Packages
ash
3.27.5
3.29.3
3.29.2
3.29.1
3.29.0
3.28.0
3.27.8
3.27.7
3.27.6
3.27.5
3.27.4
3.27.3
3.27.2
3.27.1
3.27.0
3.26.0
3.25.2
3.25.1
3.25.0
retired
3.24.7
3.24.6
3.24.5
3.24.4
3.24.3
3.24.2
3.24.1
3.24.0
3.23.1
3.23.0
3.22.2
3.22.1
3.22.0
3.21.3
3.21.2
3.21.1
3.21.0
3.20.0
3.19.3
3.19.2
3.19.1
3.19.0
3.18.0
3.17.1
3.17.0
3.16.0
3.15.0
3.14.1
3.14.0
retired
3.13.2
3.13.1
3.13.0
3.12.0
3.11.3
3.11.2
3.11.1
3.11.0
3.10.1
3.10.0
3.9.0
3.8.0
3.7.6
3.7.5
3.7.4
3.7.3
3.7.2
3.7.1
3.7.0
retired
3.6.3
retired
3.6.2
3.6.1
3.6.0
3.5.43
3.5.42
3.5.41
3.5.40
3.5.39
3.5.38
3.5.37
3.5.36
3.5.35
3.5.34
3.5.33
3.5.32
3.5.31
3.5.30
3.5.29
3.5.28
3.5.27
3.5.26
3.5.25
3.5.24
3.5.23
3.5.22
3.5.21
3.5.20
3.5.19
3.5.18
3.5.17
3.5.16
3.5.15
3.5.14
3.5.13
3.5.12
3.5.11
3.5.10
3.5.9
3.5.8
3.5.7
3.5.6
3.5.5
3.5.4
3.5.3
3.5.2
3.5.1
3.5.0
3.4.74
retired
3.4.73
3.4.72
3.4.71
3.4.70
3.4.69
3.4.68
3.4.67
3.4.66
3.4.65
3.4.64
3.4.63
3.4.62
3.4.61
3.4.60
3.4.59
3.4.58
3.4.57
3.4.56
3.4.55
3.4.54
3.4.53
3.4.52
3.4.51
3.4.50
3.4.49
3.4.48
3.4.47
3.4.46
3.4.45
3.4.44
3.4.43
3.4.42
3.4.41
3.4.40
3.4.39
3.4.38
3.4.37
3.4.36
3.4.35
3.4.34
3.4.33
3.4.32
3.4.31
3.4.30
3.4.29
3.4.28
3.4.27
3.4.26
3.4.25
3.4.24
3.4.23
3.4.22
3.4.21
3.4.20
3.4.19
3.4.18
3.4.17
3.4.16
3.4.15
3.4.14
3.4.13
3.4.12
3.4.11
3.4.10
3.4.9
3.4.8
3.4.7
3.4.6
3.4.5
3.4.4
3.4.3
3.4.2
3.4.1
3.4.0
3.3.3
3.3.2
3.3.1
3.3.0
3.2.6
3.2.5
3.2.4
3.2.3
3.2.2
3.2.1
3.2.0
3.1.8
3.1.7
3.1.6
3.1.5
3.1.4
3.1.3
3.1.2
3.1.1
3.1.0
3.0.16
3.0.15
3.0.14
3.0.13
3.0.12
3.0.11
3.0.10
3.0.9
3.0.8
3.0.7
3.0.6
3.0.5
3.0.4
3.0.3
3.0.2
3.0.1
3.0.0
3.0.0-rc.46
3.0.0-rc.45
3.0.0-rc.44
3.0.0-rc.43
3.0.0-rc.42
3.0.0-rc.41
3.0.0-rc.40
3.0.0-rc.39
3.0.0-rc.38
3.0.0-rc.37
3.0.0-rc.36
3.0.0-rc.35
3.0.0-rc.34
3.0.0-rc.33
3.0.0-rc.32
3.0.0-rc.31
3.0.0-rc.29
3.0.0-rc.28
3.0.0-rc.27
3.0.0-rc.26
3.0.0-rc.25
3.0.0-rc.24
3.0.0-rc.23
3.0.0-rc.22
3.0.0-rc.21
3.0.0-rc.20
3.0.0-rc.19
3.0.0-rc.18
3.0.0-rc.17
3.0.0-rc.16
3.0.0-rc.15
3.0.0-rc.14
3.0.0-rc.13
3.0.0-rc.12
3.0.0-rc.11
3.0.0-rc.10
3.0.0-rc.9
3.0.0-rc.8
3.0.0-rc.7
3.0.0-rc.6
3.0.0-rc.5
3.0.0-rc.4
3.0.0-rc.3
3.0.0-rc.1
3.0.0-rc.0
2.21.15
2.21.14
2.21.13
2.21.12
2.21.11
2.21.10
2.21.9
2.21.8
2.21.7
2.21.6
2.21.5
2.21.4
2.21.3
2.21.2
2.21.1
2.21.0
2.20.3
2.20.2
2.20.1
2.20.0
2.19.14
2.19.13
2.19.12
2.19.11
2.19.10
2.19.9
2.19.8
2.19.7
2.19.6
2.19.5
2.19.4
2.19.3
retired
2.19.2
retired
2.19.1
retired
2.19.0
retired
2.18.2
2.18.1
2.18.0
2.17.24
2.17.23
2.17.22
2.17.21
2.17.20
2.17.19
2.17.18
2.17.17
2.17.16
2.17.15
2.17.14
2.17.13
2.17.12
2.17.11
2.17.10
2.17.9
2.17.8
2.17.7
2.17.6
2.17.5
2.17.4
2.17.3
2.17.2
2.17.1
2.17.0
2.16.1
2.16.0
2.15.20
2.15.19
2.15.18
2.15.17
2.15.16
2.15.15
2.15.14
2.15.13
2.15.12
2.15.11
2.15.10
2.15.9
2.15.8
2.15.7
2.15.6
2.15.5
2.15.4
2.15.2
2.15.1
2.15.0
2.14.21
2.14.20
2.14.19
2.14.18
2.14.17
2.14.16
2.14.15
2.14.14
2.14.13
2.14.12
2.14.11
2.14.10
2.14.9
2.14.8
2.14.7
2.14.6
2.14.5
2.14.4
2.14.3
2.14.2
2.14.1
2.14.0
2.13.4
retired
2.13.3
2.13.2
2.13.1
2.13.0
2.12.1
2.12.0
2.11.11
2.11.10
2.11.9
2.11.8
2.11.7
2.11.6
2.11.5
2.11.4
2.11.3
2.11.2
2.11.1
2.11.0
2.11.0-rc.3
2.11.0-rc.2
2.11.0-rc.1
2.11.0-rc.0
2.10.2
2.10.1
2.10.0
2.9.29
2.9.28
2.9.27
2.9.26
2.9.25
2.9.24
2.9.23
2.9.22
2.9.21
2.9.20
2.9.19
2.9.18
2.9.17
2.9.16
2.9.15
2.9.14
2.9.13
2.9.12
2.9.11
2.9.10
2.9.9
2.9.8
2.9.7
2.9.6
2.9.5
2.9.4
2.9.3
2.9.2
2.9.1
2.9.0
2.8.1
2.8.0
2.7.1
2.7.0
2.6.31
2.6.30
2.6.29
2.6.28
2.6.27
2.6.26
2.6.25
2.6.24
2.6.23
2.6.22
2.6.21
2.6.20
2.6.19
2.6.18
2.6.17
2.6.16
2.6.15
2.6.14
2.6.13
2.6.11
2.6.10
2.6.9
2.6.8
2.6.7
2.6.6
2.6.5
2.6.4
2.6.3
2.6.2
2.6.1
2.6.0
2.5.16
2.5.15
2.5.14
2.5.13
2.5.12
2.5.11
2.5.10
2.5.9
2.5.8
2.5.7
2.5.6
2.5.5
2.5.4
2.5.3
2.5.2
2.5.1
2.5.0
2.5.0-rc.6
2.5.0-rc.5
2.5.0-rc.4
2.5.0-rc.3
2.5.0-rc.2
2.5.0-rc.1
2.5.0-rc.0
2.4.30
2.4.29
2.4.28
2.4.27
2.4.26
2.4.25
2.4.24
2.4.23
2.4.22
2.4.21
2.4.20
2.4.19
2.4.18
2.4.17
2.4.16
2.4.15
2.4.14
2.4.13
2.4.12
2.4.11
2.4.10
2.4.9
2.4.8
2.4.7
2.4.6
2.4.5
2.4.4
2.4.3
2.4.2
2.4.1
2.4.0
2.3.0
2.2.0
2.1.0
2.0.0
2.0.0-rc.15
2.0.0-rc.14
2.0.0-rc.13
2.0.0-rc.12
2.0.0-rc.11
2.0.0-rc.10
2.0.0-rc.9
2.0.0-rc.8
2.0.0-rc.7
2.0.0-rc.6
2.0.0-rc.5
2.0.0-rc.4
2.0.0-rc.3
2.0.0-rc.2
2.0.0-rc.1
2.0.0-rc.0
2.0.0-pre.8
2.0.0-pre.7
2.0.0-pre.6
2.0.0-pre.5
2.0.0-pre.4
2.0.0-pre.3
2.0.0-pre.2
2.0.0-pre.1
2.0.0-pre.0
1.53.3
1.53.2
1.53.0
1.52.0-rc.22
1.52.0-rc.21
1.52.0-rc.20
1.52.0-rc.19
1.52.0-rc.18
1.52.0-rc.17
1.52.0-rc.16
1.52.0-rc.15
1.52.0-rc.14
1.52.0-rc.13
1.52.0-rc.12
1.52.0-rc.11
1.52.0-rc.10
1.52.0-rc.9
1.52.0-rc.8
1.52.0-rc.7
1.52.0-rc.6
1.52.0-rc.5
1.52.0-rc.4
1.52.0-rc.3
1.52.0-rc.2
1.52.0-rc.1
1.52.0-rc.0
1.51.2
1.51.1
retired
1.51.0
1.50.21
1.50.20
1.50.19
1.50.18
1.50.17
1.50.16
1.50.15
1.50.14
1.50.13
1.50.12
1.50.11
1.50.10
1.50.9
1.50.8
1.50.7
1.50.6
1.50.5
1.50.4
1.50.3
1.50.2
1.50.1
1.50.0
1.49.0
1.48.0-rc.30
1.48.0-rc.29
1.48.0-rc.28
1.48.0-rc.27
1.48.0-rc.26
1.48.0-rc.25
1.48.0-rc.24
1.48.0-rc.23
1.48.0-rc.22
1.48.0-rc.21
1.48.0-rc.20
1.48.0-rc.19
1.48.0-rc.18
1.48.0-rc.17
1.48.0-rc.16
1.48.0-rc.15
1.48.0-rc.14
1.48.0-rc.13
1.48.0-rc.12
1.48.0-rc.11
1.48.0-rc.10
1.48.0-rc.9
1.48.0-rc.8
1.48.0-rc.7
1.48.0-rc.6
1.48.0-rc.5
1.48.0-rc.4
1.48.0-rc.3
1.48.0-rc.2
1.48.0-rc.1
1.48.0-rc.0
1.47.12
1.47.11
1.47.10
1.47.9
1.47.8
1.47.7
1.47.6
1.47.5
1.47.4
1.47.3
1.47.2
1.47.1
1.47.0
1.46.13
1.46.12
1.46.11
1.46.10
1.46.9
1.46.8
1.46.7
1.46.6
1.46.5
1.46.4
1.46.3
1.46.2
1.46.1
1.46.0
1.45.0-rc9
1.45.0-rc8
1.45.0-rc7
1.45.0-rc6
1.45.0-rc5
1.45.0-rc4
1.45.0-rc3
1.45.0-rc20
1.45.0-rc2
1.45.0-rc19
1.45.0-rc18
1.45.0-rc17
1.45.0-rc16
1.45.0-rc15
1.45.0-rc14
1.45.0-rc13
1.45.0-rc12
1.45.0-rc11
1.45.0-rc10
1.45.0-rc1
1.45.0-rc0
1.44.13
1.44.12
1.44.11
1.44.10
1.44.9
1.44.8
1.44.7
1.44.6
1.44.5
1.44.4
1.44.3
1.44.2
1.44.1
1.44.0
1.43.12
1.43.11
1.43.10
1.43.9
1.43.8
1.43.7
1.43.6
1.43.5
1.43.4
1.43.3
1.43.2
1.43.1
1.43.0
1.42.0
1.41.12
1.41.11
1.41.10
1.41.9
1.41.8
1.41.7
1.41.6
1.41.5
1.41.4
1.41.3
1.41.2
1.41.1
1.41.0
1.40.0
1.39.7
1.39.6
1.39.5
1.39.4
1.39.3
1.39.2
1.39.1
1.39.0
1.38.0
1.37.2
1.37.1
1.37.0
1.36.22
1.36.21
1.36.19
1.36.18
1.36.17
1.36.16
1.36.15
1.36.14
1.36.13
1.36.12
1.36.11
1.36.10
1.36.9
1.36.8
1.36.7
1.36.6
1.36.5
1.36.4
1.36.3
1.36.2
1.36.0
1.35.1
1.35.0
1.34.9
1.34.8
1.34.7
1.34.6
1.34.5
1.34.4
1.34.3
1.34.2
1.34.1
1.34.0
1.33.0
1.32.2
1.32.1
1.32.0
1.31.1
1.31.0
1.30.2
1.30.1
1.29.0-rc1
1.29.0-rc0
1.28.1
1.28.0
1.27.1
1.27.0
1.26.13
1.26.12
1.26.11
1.26.10
1.26.9
1.26.8
1.26.7
1.26.6
1.26.5
1.26.4
1.26.2
1.26.1
1.26.0
1.25.8
1.25.7
1.25.6
1.25.5
1.25.4
1.25.3
1.25.2
1.25.1
1.25.0
1.24.2
1.24.1
1.24.0
1.23.3
1.23.2
1.23.1
1.23.0
1.22.1
1.22.0
1.20.1
1.20.0
1.19.1
1.19.0
1.18.1
1.18.0
1.17.1
1.17.0
1.16.2
1.15.1
1.15.0
1.14.0
1.13.4
1.13.3
1.13.2
1.13.1
1.13.0
1.12.0
1.11.1
1.11.0
1.10.0
1.9.0
1.8.0
1.7.0
1.6.8
1.6.7
1.6.6
1.6.5
1.6.4
1.6.3
1.6.2
1.6.1
1.6.0
1.5.1
1.5.0
1.4.1
1.4.0
1.3.1
1.3.0
1.2.1
1.2.0
1.1.3
1.1.2
1.1.0
1.0.3
1.0.2
1.0.1
1.0.0
0.13.1
0.13.0
0.12.0
0.10.0
0.9.1
0.9.0
0.8.0
0.7.0
0.6.5
0.6.4
0.6.3
0.6.2
0.6.1
0.6.0
0.5.2
0.5.1
0.5.0
0.4.0
0.3.0
0.2.0
0.1.9
0.1.8
0.1.3
0.1.1
0.1.0
A declarative, extensible framework for building Elixir applications.
Security advisory:
This version has known vulnerabilities.
View advisories
Current section
Files
Jump to
Current section
Files
lib/ash/notifier/pub_sub/pub_sub.ex
# SPDX-FileCopyrightText: 2019 ash contributors <https://github.com/ash-project/ash/graphs/contributors>
#
# SPDX-License-Identifier: MIT
defmodule Ash.Notifier.PubSub do
require Logger
@publish %Spark.Dsl.Entity{
name: :publish,
target: Ash.Notifier.PubSub.Publication,
transform: {Ash.Notifier.PubSub.Publication, :transform, []},
describe: "Configure a given action to publish its results over a given topic.",
examples: [
"publish :create, \"created\"",
"publish \"my_event\", :create, \"created\"",
"""
publish :assign, "assigned"
"""
],
schema: Ash.Notifier.PubSub.Publication.schema(),
args: [{:optional, :event}, :action, :topic]
}
@publish_all %Spark.Dsl.Entity{
name: :publish_all,
target: Ash.Notifier.PubSub.Publication,
transform: {Ash.Notifier.PubSub.Publication, :transform, []},
describe: """
Works the same as `publish`, except that it takes a type and publishes all actions of that type.
""",
examples: [
"publish_all :create, \"created\"",
"publish_all \"my_creates\", :create, \"created\""
],
schema: Ash.Notifier.PubSub.Publication.publish_all_schema(),
args: [{:optional, :event}, :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)}`
## Loading data before broadcast
Publications support a `load` option that declares fields to load onto `notification.data`
before the notification is broadcast. This is useful when a `transform` or `filter` function,
or a subscriber, needs access to related data that is not present on the record returned by
the action.
The value is any load statement accepted by `Ash.Query.load/2`.
```elixir
pub_sub do
module MyAppWeb.Endpoint
prefix "post"
publish :create, "created", load: [:comment, author: [:full_name]]
publish :update, ["updated", :id], load: [:comment, author: [:full_name]]
end
```
Different publications on the same resource can request different loads. Loads that are
identical across publications are deduplicated and fetched only once.
"""
use Spark.Dsl.Extension,
sections: @sections,
persisters: [
Ash.Notifier.PubSub.Persisters.SetCalcReturns
],
verifiers: [
Ash.Notifier.PubSub.Verifiers.VerifyActionNames,
Ash.Notifier.PubSub.Verifiers.VerifyCalcTransforms
]
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))
|> Enum.with_index()
|> Enum.each(fn {pub, index} ->
if (is_nil(pub.filter) || pub.filter.(notification)) &&
(is_nil(filter) || filter.(notification)) do
enriched = enrich_for_publication(notification, pub, index)
publish_notification(pub, enriched)
end
end)
end
@doc false
def load(resource, action) do
resource
|> Ash.Notifier.PubSub.Info.publications()
|> Enum.filter(&matches?(&1, action))
|> Enum.with_index()
|> Enum.flat_map(fn {publication, index} ->
calc_loads = load_for_calc_transform(publication, resource)
statement_loads =
case List.wrap(publication.load) do
[] ->
[]
statement ->
case Ash.Query.Calculation.new(
{:pub_sub_dependency, index},
Ash.Notifier.NotifierDependencies,
[statement: statement],
Ash.Type.Map,
[]
) do
{:ok, calc} -> [calc]
_ -> []
end
end
calc_loads ++ statement_loads
end)
end
defp load_for_calc_transform(%{transform: calc_name}, resource)
when is_atom(calc_name) and not is_nil(calc_name) do
case Ash.Resource.Info.calculation(resource, calc_name) do
nil ->
[]
calc ->
case Ash.Query.Calculation.new(
calc_name,
elem(calc.calculation, 0),
elem(calc.calculation, 1),
calc.type,
calc.constraints
) do
{:ok, query_calc} -> [query_calc]
_ -> []
end
end
end
defp load_for_calc_transform(_, _), do: []
# Enriches a notification with the per-publication loaded data before broadcasting.
defp enrich_for_publication(notification, pub, index) do
case List.wrap(pub.load) do
[] ->
notification
statement ->
extra =
get_in(notification.data, [Access.key(:calculations, %{}), {:pub_sub_dependency, index}]) ||
%{}
resource = notification.resource
enriched_data =
Enum.reduce(extra, notification.data, fn {key, value}, data ->
case Ash.Notifier.placement_for_key(key, statement, resource) do
:struct -> Map.put(data, key, value)
:calculations -> Map.update!(data, :calculations, &Map.put(&1 || %{}, key, value))
end
end)
%{notification | data: enriched_data}
end
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 = to_string(publish.event || 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 =
cond do
is_atom(transform) and not is_nil(transform) ->
get_calculation_value(notification, transform)
is_function(transform) ->
transform.(notification)
true ->
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
defp get_calculation_value(notification, calc_name) do
Map.get(notification.data.calculations, calc_name)
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