defmodule AshDispatch.Dispatcher do @moduledoc """ Core dispatcher that handles event delivery across all transports. This module orchestrates the dispatch process: 1. Creates DeliveryReceipt with full content 2. Routes to appropriate transport handler 3. Updates receipt status based on delivery result ## Receipt-First Pattern All deliveries follow the receipt-first pattern: - Receipt created BEFORE delivery attempt - Full content stored in receipt (enables retries without re-rendering) - Status tracked through lifecycle: pending → scheduled/sent → failed ## Transport Routing - `:in_app` → `AshDispatch.Transports.InApp` (synchronous) - `:email` → `AshDispatch.Transports.Email` (async via Oban) - `:discord` → `AshDispatch.Transports.Discord` (async via Oban) - `:slack` → `AshDispatch.Transports.Slack` (async via Oban) - `:sms` → `AshDispatch.Transports.SMS` (async via Oban) - `:webhook` → `AshDispatch.Transports.Webhook` (async via Oban) - `:broadcast` → `AshDispatch.Transports.Broadcast` (lightweight, no receipt) - `:oban` → `AshDispatch.Transports.Oban` (lightweight, enqueues arbitrary worker named in event `:metadata`; no receipt) """ alias AshDispatch.Config alias AshDispatch.Context alias AshDispatch.ChannelResolver alias AshDispatch.EventResolver alias AshDispatch.Naming alias AshDispatch.Event.RecipientExtractor alias AshDispatch.Helpers.RecordReader require Logger @doc """ Dispatches an event by ID with data and variables. This is the high-level dispatch function that applications use. It looks up the event module, creates channels, and dispatches to all of them. ## Parameters - `event_id` - The event identifier (e.g., "requests.new_reseller_request") - `data` - Map of domain data (resources, users, etc.) - `variables` - Map of template variables (tokens, simple values, etc.) - defaults to %{} ## Returns - `{:ok, results}` - List of delivery receipt results - `{:error, reason}` - If event not found or dispatch fails ## Data vs Variables Use `data` for: - Domain resources (User, Order, Ticket, etc.) - Structured Ash resources - Objects that need relationship traversal Use `variables` for: - Authentication tokens (reset_token, confirmation_token) - Simple template values - Computed values that don't need relationships This separation prevents naming conflicts and makes the intent clearer. ## Examples # Dispatch with authentication token AshDispatch.Dispatcher.dispatch( "accounts.password_reset", %{user: user}, %{reset_token: token} ) # Dispatch with invitation data AshDispatch.Dispatcher.dispatch( "accounts.invited", %{invited_user: user, invited_by: admin}, %{invitation_token: token, custom_message: message} ) # Dispatch an order created event (no variables needed) AshDispatch.Dispatcher.dispatch( "orders.created", %{order: order, user: user} ) """ # Function head with default parameter def dispatch(event_id, data, variables \\ %{}) # Module-typed overload — resolve event_id via EventRegistry reverse-lookup. # Eliminates hand-typed event_id strings at call sites (typo risk). Forge.Binding # 019e2795-afd4 (Mosis project) — surfaced during predict_incremental wiring # where the dispatch ID `"admin_events.corpus_fill_completed"` was typed by hand # at the emission site with no compile-time validation against the DSL. # # Usage: # AshDispatch.Dispatcher.dispatch(MyApp.Events.OrderCreated, %{...}, %{...}) # Equivalent to: # AshDispatch.Dispatcher.dispatch(MyApp.Events.OrderCreated.id() or registry-id, ...) def dispatch(module, data, variables) when is_atom(module) and is_map(data) and is_map(variables) do case resolve_event_id_for_module(module) do {:ok, event_id} -> dispatch(event_id, data, variables) :error -> {:error, {:event_module_not_registered, "module #{inspect(module)} is not registered in any Ash domain's dispatch DSL — " <> "either register it via `event(:x, module: #{inspect(module)})` or use the string-id form."}} end end def dispatch(event_id, data, variables) when is_binary(event_id) and is_map(data) and is_map(variables) do # Use centralized EventResolver for consistent event lookup case EventResolver.find_module(event_id) do {:ok, event_module} -> # Create initial context (before generate_send_variables) initial_context = %Context{ event_id: event_id, data: data, variables: variables, user: extract_user_from_data(data) } # Generate real variables using EventResolver (handles function_exported? and error handling) # This allows events to provide real data (tokens, etc.) for actual sending # while using sample_data() for previews enhanced_vars_result = EventResolver.generate_send_variables(event_module, initial_context, variables) case enhanced_vars_result do {:ok, enhanced_variables} -> # Update context with enhanced variables context = %{initial_context | variables: enhanced_variables} # Use centralized ChannelResolver for consistent priority logic channels = ChannelResolver.resolve(event_id, event_module, context) # Build event config — pull the inline event's :metadata + # :content into event_config so transports that depend on # them (e.g. AshDispatch.Transports.Oban needs # event_config[:metadata][:oban_worker]) read the substrate # the DSL declared. Pre-fix, event_config only carried # :module, so :oban transport always logged # "missing oban_worker in metadata" and silently skipped # the enqueue. ω.13.swarm-2c103-test-bugs. event_config = case AshDispatch.EventRegistry.find_event(event_id) do {:ok, event_info} -> %{ module: event_module, metadata: event_info[:metadata] || [], content: event_info[:content] || [] } {:error, _} -> %{module: event_module} end # Dispatch to all channels # Deduplication is opt-in via deduplicate_in_app: true in event config results = dispatch_all_channels(channels, context, event_config) # Return success if any dispatch succeeded if Enum.any?(results, fn {:ok, _} -> true _ -> false end) do {:ok, results} else {:error, :all_dispatches_failed} end {:error, reason} -> Logger.error("generate_send_variables failed for #{event_id}: #{inspect(reason)}") {:error, reason} end {:error, :not_found} -> Logger.error("Event module not found for event_id: #{event_id}") {:error, :event_not_found} end end @doc """ Low-level dispatch function that dispatches to a specific channel with a pre-built context. This is used internally and by DSL-based events that build their own context. Most applications should use the high-level `dispatch/2` or `dispatch/3` functions instead. ## Parameters - `context` - Pre-built AshDispatch.Context struct - `channel` - AshDispatch.Channel struct - `event_config` - Event configuration map (with :module, :content, etc.) ## Returns - `{:ok, receipt}` - DeliveryReceipt if successful - `{:error, reason}` - If dispatch fails """ def dispatch_channel(context, channel, event_config) do # Apply channel-level load (additional to event-level load) context = apply_channel_load(context, channel) # Resolve recipients for this channel recipients = resolve_recipients_for_channel(context, channel, event_config) # Filter out actor if exclude_actor is set recipients = filter_out_actor(recipients, context, channel) # Create one receipt per recipient results = Enum.map(recipients, fn recipient -> dispatch_to_recipient(context, channel, event_config, recipient) end) # Return success if at least one dispatch succeeded successful = Enum.filter(results, fn {:ok, _} -> true _ -> false end) if length(successful) > 0 do {:ok, Enum.map(successful, fn {:ok, receipt} -> receipt end)} else {:error, :all_dispatches_failed} end end # Dispatch to all channels with optional deduplication based on deduplicate_group # Channels with the same deduplicate_group are deduplicated - user gets max one notification per group defp dispatch_all_channels(channels, context, event_config) do # Check if any channels have deduplicate_group set has_dedup_groups? = Enum.any?(channels, & &1.deduplicate_group) if has_dedup_groups? do dispatch_with_deduplication(channels, context, event_config) else # No deduplication - dispatch each channel independently (default) Enum.map(channels, fn channel -> dispatch_channel(context, channel, event_config) end) end end # Dispatch channels with deduplication based on deduplicate_group defp dispatch_with_deduplication(channels, context, event_config) do # Collect all recipients from all channels with their channel info recipients_with_channels = Enum.flat_map(channels, fn channel -> channel_context = apply_channel_load(context, channel) recipients = resolve_recipients_for_channel(channel_context, channel, event_config) # Filter out actor if exclude_actor is set for this channel recipients = filter_out_actor(recipients, channel_context, channel) Enum.map(recipients, fn recipient -> {recipient, channel} end) end) # Apply deduplication - users in the same deduplicate_group get max one notification deduplicated_recipients = apply_group_deduplication(recipients_with_channels) # Dispatch to each recipient using their associated channel Enum.map(deduplicated_recipients, fn {recipient, channel} -> channel_context = apply_channel_load(context, channel) dispatch_to_recipient(channel_context, channel, event_config, recipient) end) end # Apply deduplication based on channel's deduplicate_group # Channels with same group are deduplicated - first channel (by DSL order) wins # Channels without a group are never deduplicated defp apply_group_deduplication(recipients_with_channels) do # Track which user IDs we've seen per dedup group {result, _seen} = Enum.reduce(recipients_with_channels, {[], %{}}, fn {recipient, channel} = item, {acc, seen} -> user_id = get_user_id(recipient) dedup_group = channel.deduplicate_group cond do # No dedup group - always include is_nil(dedup_group) -> {[item | acc], seen} # Check if we've already seen this user in this dedup group MapSet.member?(Map.get(seen, dedup_group, MapSet.new()), user_id) -> # Skip - user already received notification for this group {acc, seen} true -> # First time seeing this user in this group - include and mark as seen new_seen = Map.update(seen, dedup_group, MapSet.new([user_id]), &MapSet.put(&1, user_id)) {[item | acc], new_seen} end end) Enum.reverse(result) end # Private function that dispatches to a specific recipient defp dispatch_to_recipient(context, channel, event_config, recipient) do # Skip receipt creation for lightweight transports (e.g., :broadcast) if skip_receipt_for_transport?(channel.transport) do dispatch_without_receipt(context, channel, event_config, recipient) else dispatch_with_receipt(context, channel, event_config, recipient) end end # F1 — receipt-skip is now declared on each Transport via the # `use AshDispatch.Transport, skip_receipt?: bool` macro. Lookup # routes through `AshDispatch.Transport.Registry` so adding a # lightweight transport (broadcast / oban / future) is a single # `use` line — no Dispatcher edit needed. defp skip_receipt_for_transport?(transport_atom), do: AshDispatch.Transport.Registry.skip_receipt?(transport_atom) defp extract_user_id_from_recipient(%{id: id}), do: id defp extract_user_id_from_recipient(%{"id" => id}), do: id defp extract_user_id_from_recipient(_), do: nil defp dispatch_without_receipt(context, channel, event_config, recipient) do # Build a minimal pseudo-receipt for the transport (no DB write) pseudo_receipt = %{ id: nil, event_id: context.event_id, transport: channel.transport, audience: channel.audience, user_id: extract_user_id_from_recipient(recipient), content: %{}, status: :pending, metadata: %{} } case dispatch_to_transport(pseudo_receipt, context, channel, event_config) do {:ok, _} -> Logger.debug("Broadcast dispatched: #{context.event_id} to #{channel.audience}") {:ok, :broadcast_sent} {:error, reason} -> Logger.warning("Broadcast failed: #{context.event_id} — #{inspect(reason)}") {:error, reason} end end defp dispatch_with_receipt(context, channel, event_config, recipient) do case create_receipt(context, channel, event_config, recipient) do {:ok, :skipped_optional} -> # Optional channel that can't be delivered (e.g. SMS recipient # without a phone number). Counted as success — the channel # opted into "best-effort" delivery via `optional: true`. {:ok, :skipped_optional} {:ok, receipt} -> case dispatch_to_transport(receipt, context, channel, event_config) do {:ok, updated_receipt} -> Logger.debug(""" Event dispatched successfully Event: #{context.event_id} Transport: #{channel.transport} Audience: #{channel.audience} Recipient: #{inspect(recipient)} Receipt ID: #{updated_receipt.id} Status: #{updated_receipt.status} """) # Broadcast counters if configured (once per recipient) broadcast_counters(context, channel, event_config) {:ok, updated_receipt} {:error, error} -> {:error, error} end {:error, error} -> {:error, error} end end # Private functions defp resolve_recipients_for_channel(context, channel, event_config) do # Resolve module with runtime fallback (handles compilation order issues) module = resolve_event_module(event_config, context) # Priority order for recipient resolution: # 1. recipient_resolver config (new DSL-based system) # 2. Event module recipients/2 callback (if returns non-empty) # 3. Legacy audience config (MFA-based) cond do # New: Use configured recipient resolver module resolver = Config.recipient_resolver() -> resource = extract_primary_resource(context) resolver.resolve(channel.audience, resource, context) # If there's a recipient_filter in event_config, use it (inline DSL or event-level config) not is_nil(event_config[:recipient_filter]) -> AshDispatch.Event.Helpers.resolve_recipients_for_audience(context, channel, event_config) # If there's a module, try its recipients callback first # But fall back to audience-based resolution if it returns empty not is_nil(module) -> case EventResolver.recipients(module, context, channel) do # Module returned recipients - use them recipients when is_list(recipients) and recipients != [] -> recipients # Module returned empty list - fall back to audience-based resolution # This handles both explicit empty returns and default [] from call_if_exported _ -> AshDispatch.Event.Helpers.resolve_recipients_for_audience( context, channel, event_config ) end # Pure inline DSL without module - use helpers with app-level config true -> AshDispatch.Event.Helpers.resolve_recipients_for_audience(context, channel, event_config) end end # Extract the primary resource from context for recipient resolution defp extract_primary_resource(context) do # Priority order: # 1. Use resource_key if specified # 2. Look for common resource keys in data # 3. Return first struct found in data cond do context.resource_key && Map.has_key?(context.data, context.resource_key) -> Map.get(context.data, context.resource_key) true -> # Try common resource keys, then fall back to first struct common_keys = [:project, :lead, :order, :meeting, :business_plan, :offer, :phase] Enum.find_value(common_keys, fn key -> case Map.get(context.data, key) do value when is_struct(value) -> value _ -> nil end end) || context.data |> Map.values() |> Enum.find(&is_struct/1) end end defp create_receipt(context, channel, event_config, recipient) do # Build full content for the receipt. Recipient is threaded in so # per-recipient locale resolution (recipient.locale) can flip the # rendered subject/body/text per recipient — multi-recipient channels # no longer share a single pre-rendered content blob when locales # differ. content = build_receipt_content(context, channel, event_config, recipient) # Extract recipient identifier and name using configured fields. # `:skip` is the soft-fail return when `channel.optional: true` and # the identifier can't be extracted — propagate up so the caller # skips this channel without crashing the whole dispatch. case extract_recipient_identifier(recipient, channel, event_config) do :skip -> {:ok, :skipped_optional} recipient_identifier -> do_create_receipt(context, channel, event_config, recipient, content, recipient_identifier) end end defp do_create_receipt(context, channel, event_config, recipient, content, recipient_identifier) do recipient_name = extract_recipient_name(recipient, channel, event_config) recipient_user_id = get_user_id(recipient) # Always link email receipts to in-app notifications for the same user/event # This enables skip_if_read policy and provides useful linking for analytics notification_id = if channel.transport == :email and recipient_user_id do find_in_app_notification_id(context.event_id, recipient_user_id) else nil end # Extract source resource info for linking (e.g., link receipt to order) {source_type, source_id} = extract_source_info(context, event_config) # Resolve locale for traceability (which template was rendered). # Per-recipient locale (recipient.locale) is considered so the receipt # records the actual locale that was used to render the recipient's # content blob. locale = resolve_channel_locale(channel, context, recipient) # Build receipt attributes attrs = %{ event_id: context.event_id, transport: channel.transport, audience: channel.audience, recipient: recipient_identifier, user_id: recipient_user_id, notification_id: notification_id, source_type: source_type, source_id: source_id, locale: locale # Note: scheduled_for removed - Oban handles scheduling via schedule_in parameter } # Add recipient name to content if available content = if recipient_name do Map.put(content, :recipient_name, recipient_name) else content end # Add transport-specific content fields attrs = case channel.transport do :email -> Map.merge(attrs, %{ subject: content[:subject], body_html: content[:html_body], body_text: content[:text_body], content: content }) _ -> Map.put(attrs, :content, content) end # Create DeliveryReceipt # Use authorize?: false and skip_unknown_inputs to work within transaction context Config.delivery_receipt_resource() |> Ash.Changeset.for_create(:create, attrs) |> Ash.create(authorize?: false, skip_unknown_inputs: [:notification_id]) end # Extract recipient identifier using RecipientExtractor with cascading config. # When the channel is marked `optional: true` and the identifier can't be # extracted (e.g. SMS channel where the user has no phone_number), we # log + return `:skip` instead of crashing the whole dispatch — the # caller skips the receipt creation for this channel only. defp extract_recipient_identifier(recipient, channel, event_config) do RecipientExtractor.extract_identifier( recipient, channel.transport, channel.audience, event_config[:recipient] ) rescue error -> if Map.get(channel, :optional, false) do Logger.info( "Skipping optional channel — recipient has no #{channel.transport} identifier " <> "(transport=#{channel.transport} audience=#{channel.audience})" ) :skip else Logger.error(""" Failed to extract recipient identifier Transport: #{channel.transport} Audience: #{channel.audience} Recipient: #{inspect(recipient)} Error: #{inspect(error)} """) reraise error, __STACKTRACE__ end end # Extract recipient name using RecipientExtractor (returns nil if not configured) defp extract_recipient_name(recipient, channel, event_config) do RecipientExtractor.extract_name( recipient, channel.transport, channel.audience, event_config[:recipient] ) rescue error -> Logger.warning(""" Failed to extract recipient name (continuing with nil) Transport: #{channel.transport} Audience: #{channel.audience} Recipient: #{inspect(recipient)} Error: #{inspect(error)} """) nil end # Get user_id from recipient struct defp get_user_id(recipient) when is_map(recipient) do Map.get(recipient, :id) end defp get_user_id(_), do: nil # Filter out the actor (user who triggered the event) from recipients # Only applies when channel.exclude_actor is true and context.user is set defp filter_out_actor(recipients, %{user: %{id: actor_id}}, %{exclude_actor: true}) when not is_nil(actor_id) do Enum.reject(recipients, fn recipient -> get_user_id(recipient) == actor_id end) end defp filter_out_actor(recipients, _context, _channel), do: recipients # Extract source resource info (type and ID) for linking receipts to source resources # Returns {source_type, source_id} tuple where values may be nil defp extract_source_info(context, event_config) do # Resolve module with runtime fallback (handles compilation order issues) module = resolve_event_module(event_config, context) cond do # Module-based events: use resource() callback to get the source type not is_nil(module) and not is_nil(EventResolver.resource(module)) -> resource_module = EventResolver.resource(module) resource_key = get_resource_key(module, context) case Map.get(context.data, resource_key) do %{id: id} = _record when not is_nil(id) -> {to_string(resource_module), id} _ -> {to_string(resource_module), nil} end # Inline DSL events: use resource_module from event_config if present not is_nil(event_config[:resource_module]) -> resource_module = event_config[:resource_module] resource_key = context.resource_key case Map.get(context.data, resource_key) do %{id: id} = _record when not is_nil(id) -> {to_string(resource_module), id} _ -> {to_string(resource_module), nil} end # No source info available true -> {nil, nil} end end # Get resource key from module callback or context defp get_resource_key(module, context) do # Use EventResolver for safe callback execution EventResolver.data_key(module) || context.resource_key end # Find the in-app notification ID for skip_if_read policy defp find_in_app_notification_id(event_id, user_id) do require Ash.Query case Config.delivery_receipt_resource() |> Ash.Query.filter( event_id == ^event_id and user_id == ^user_id and transport == :in_app ) |> Ash.Query.select([:notification_id]) |> Ash.Query.limit(1) |> Ash.read(authorize?: false) do {:ok, [receipt | _]} -> receipt.notification_id _ -> nil end end defp build_receipt_content(context, channel, event_config, recipient) do # Set per-recipient Gettext locale BEFORE rendering callbacks so any # `dgettext` invocations inside `prepare_template_assigns`, # `notification_title/message`, content-string interpolation, and # `EEx.eval_string` template rendering pick up the right locale. # This is the same backend referenced via the `:gettext_backend` # config — set to a no-op when not configured. # # `Gettext.put_locale/2` is process-local, so we wrap the body in # `try/after` to restore whatever locale was in effect when this # receipt started building. Without that, a worker that dispatches # to user A (locale="en") and then runs unrelated `t()` calls in # the same process would observe the leaked "en" locale until the # next explicit `put_locale`. See PR with regression test. prev_locale = current_locale() apply_recipient_locale(channel, context, recipient) try do do_build_receipt_content(context, channel, event_config, recipient) after restore_locale(prev_locale) end end defp do_build_receipt_content(context, channel, event_config, recipient) do # Resolve module with runtime fallback (handles compilation order issues) module = resolve_event_module(event_config, context) base_content = case module do nil -> # Pure inline DSL - use inline content only build_inline_content(context, channel, event_config, recipient) module -> # Hybrid mode: module + inline DSL # Step 1: Enhance context with module's prepare_template_assigns if available extra_assigns = safe_prepare_template_assigns(module, context, channel) enhanced_context = if map_size(extra_assigns) > 0 do # Add extra assigns to context variables Map.update(context, :variables, extra_assigns, fn vars -> Map.merge(vars, extra_assigns) end) else context end # Step 2: Build module content (with enhanced context) module_content = build_module_content(enhanced_context, channel, module, event_config, recipient) # Step 3: Check if there's inline DSL content to merge if has_inline_content?(event_config, channel) do # Build inline content with enhanced context (so interpolation has access to module assigns) inline_content = build_inline_content(enhanced_context, channel, event_config, recipient) # Inline DSL takes precedence over module callbacks Map.merge(module_content, inline_content) else # No inline DSL, use module content only module_content end end # Add policy to content if present on channel if channel.policy do Map.put(base_content, :policy, to_string(channel.policy)) else base_content end end # Read the current process-level Gettext locale so `build_receipt_content/4` # can restore it after rendering. Returns nil when no backend is configured # or when `Gettext` isn't loaded. defp current_locale do case Config.gettext_backend() do nil -> nil backend -> if Code.ensure_loaded?(Gettext) do try do apply(Gettext, :get_locale, [backend]) rescue _ -> nil end else nil end end end # Restore a previously-captured Gettext locale. Used in the `after` # block of `build_receipt_content/4` so per-recipient locale changes # don't leak to subsequent code paths in the same process. defp restore_locale(nil), do: :ok defp restore_locale(locale) when is_binary(locale) do case Config.gettext_backend() do nil -> :ok backend -> if Code.ensure_loaded?(Gettext) do try do apply(Gettext, :put_locale, [backend, locale]) :ok rescue _ -> :ok end else :ok end end end # Set Gettext locale to the resolved per-channel-per-recipient locale. # No-op when no backend is configured or when the resolved locale is nil. defp apply_recipient_locale(channel, context, recipient) do case Config.gettext_backend() do nil -> :ok backend -> case resolve_channel_locale(channel, context, recipient) do locale when is_binary(locale) and locale != "" -> try do apply(Gettext, :put_locale, [backend, locale]) :ok rescue _ -> :ok end _ -> :ok end end end # Check if there's inline DSL content configuration defp has_inline_content?(event_config, channel) do # Check if channel or event has content/metadata defined (channel.content && map_size(channel.content) > 0) || (channel.metadata && map_size(channel.metadata) > 0) || (event_config[:content] && length(event_config[:content]) > 0) || (event_config[:metadata] && length(event_config[:metadata]) > 0) end defp build_inline_content(context, channel, event_config, recipient) do # Prefer channel-level content/metadata, fall back to event-level # This allows both patterns: # 1. Transport-specific: channels: [[transport: :email, content: [subject: "..."]]] # 2. Shared: content: [subject: "..."], channels: [[transport: :email]] channel_content = channel.content || %{} channel_metadata = channel.metadata || %{} event_content = (event_config[:content] || %{}) |> Enum.into(%{}) event_metadata = (event_config[:metadata] || %{}) |> Enum.into(%{}) # Merge with channel-level taking precedence content_config = Map.merge(event_content, channel_content) metadata_config = Map.merge(event_metadata, channel_metadata) base_content = %{ transport: channel.transport, audience: channel.audience } # Add transport-specific content with variable interpolation transport_content = case channel.transport do :email -> # Try to render templates (convention-based or explicit path). # Recipient threaded through so locale resolution can pick # recipient.locale when present. {html_body, text_body} = render_inline_email_templates(context, channel, event_config, recipient) # Build content - only include fields with actual values (to not overwrite module callbacks in hybrid mode) %{} |> maybe_put(:from, content_config[:from_email]) |> maybe_put(:subject, interpolate(content_config[:subject], context)) |> maybe_put(:html_body, html_body) |> maybe_put(:text_body, text_body) :in_app -> # Build base content with required fields base = %{ notification_type: metadata_config[:notification_type] || :info } # Add optional fields only if they have values (to not overwrite module callbacks in hybrid mode) base |> maybe_put( :title, interpolate(content_config[:title] || content_config[:notification_title], context) ) |> maybe_put( :message, interpolate( content_config[:message] || content_config[:notification_message], context ) ) |> maybe_put(:action_url, interpolate(content_config[:action_url], context)) |> maybe_put(:action_label, interpolate(content_config[:action_label], context)) :discord -> %{ message: interpolate( content_config[:message] || content_config[:notification_message], context ), webhook_url: channel.webhook_url } :slack -> %{ message: interpolate( content_config[:message] || content_config[:notification_message], context ), webhook_url: channel.webhook_url } :sms -> %{ message: interpolate( content_config[:message] || content_config[:notification_message], context ) } :webhook -> %{ payload: content_config[:webhook_payload] || %{}, webhook_url: channel.webhook_url } end Map.merge(base_content, transport_content) end # Render email templates for inline DSL events defp render_inline_email_templates(context, channel, event_config, recipient) do # Get template configuration template_path = event_config[:template_path] event_id = context.event_id otp_app = get_otp_app(context) # Domain name for template path resolution domain = event_config[:domain] # Resource name for template path resolution resource_name = event_config[:resource_name] variant = channel.variant # Locale priority: channel.locale > channel.locale_from (dynamic) > # recipient.locale > context.locale locale = resolve_channel_locale(channel, context, recipient) # Prepare template assigns assigns = Context.template_assigns(context) # Try to render HTML template html = case AshDispatch.TemplateResolver.render( template_path: template_path, event_id: event_id, otp_app: otp_app, # Pass domain for correct path derivation domain: domain, # Pass resource_name for collision prevention resource_name: resource_name, format: :html, transport: :email, variant: variant, locale: locale, assigns: assigns ) do {:ok, rendered} -> rendered {:error, :template_not_found} -> nil {:error, error} -> Logger.warning("Failed to render HTML template for #{event_id}: #{inspect(error)}") nil end # Try to render text template text = case AshDispatch.TemplateResolver.render( template_path: template_path, event_id: event_id, otp_app: otp_app, # Pass domain for correct path derivation domain: domain, # Pass resource_name for collision prevention resource_name: resource_name, format: :text, transport: :email, variant: variant, locale: locale, assigns: assigns ) do {:ok, rendered} -> rendered {:error, :template_not_found} -> nil {:error, error} -> Logger.warning("Failed to render text template for #{event_id}: #{inspect(error)}") nil end {html, text} end # Get OTP app name from context or fallback to default defp get_otp_app(context) do # Try to extract from resource_module if available (for module-based events) resource_module = Map.get(context, :resource_module) if resource_module do case Atom.to_string(resource_module) do "Elixir." <> rest -> rest |> String.split(".") |> List.first() |> String.downcase() |> String.to_atom() _ -> :ash_dispatch end else # For inline DSL events, derive from event_id # e.g., "requests.new_reseller_request" -> extract "Magasin" from data key derive_otp_app_from_event_id(context.event_id) end end # Derive OTP app - use configured value (single source of truth) defp derive_otp_app_from_event_id(_event_id) do Config.otp_app() || :magasin end # Derive event directory from module name for file-based template loading # Uses Naming.module_directory for consistent path derivation # Example: Magasin.Accounts.Events.PasswordReset.Event -> lib/magasin/accounts/events/password_reset defp derive_event_dir_from_module(module, _otp_app) when is_atom(module) do # Use Naming for consistent path derivation relative_path = Naming.module_directory(module) # Convert to absolute path for file operations # In development/test: use source directory # In production: this won't be called (uses priv manifest instead) source_path = Path.join([File.cwd!(), relative_path]) if File.exists?(source_path), do: source_path, else: nil rescue _ -> nil end # Interpolate variables in a string template defp interpolate(nil, _context), do: nil defp interpolate(template, context) when is_binary(template) do # Step 1: Translate via Gettext if backend configured translated = translate_content(template, context.locale) # Step 2: Variable interpolation ({{variable}} → value) assigns = Context.template_assigns(context) AshDispatch.VariableInterpolator.interpolate(translated, assigns, context.resource_key) end defp interpolate(value, _context), do: value # Translate content string via Gettext if a backend is configured. # The content string is the msgid, looked up in the configured domain # (`AshDispatch.Config.gettext_domain/0`, default `"notifications"`). # Gettext calls are dynamic to avoid requiring gettext as a dependency. # # **Locale handling:** if `locale` is non-nil we set it explicitly; # otherwise we trust the process-level locale that # `apply_recipient_locale/3` already put in place (per-recipient # resolution from 0.4.5). Previously this unconditionally reset the # locale to `"en"` for nil input, which silently undid recipient-locale # resolution for DSL-content lookups. defp translate_content(string, locale) do case AshDispatch.Config.gettext_backend() do nil -> string backend -> if Code.ensure_loaded?(Gettext) do if is_binary(locale) and locale != "" do apply(Gettext, :put_locale, [backend, locale]) end apply(Gettext, :dgettext, [backend, AshDispatch.Config.gettext_domain(), string]) else string end end end # Helper to conditionally add a key-value pair to a map only if value is not nil # This prevents overwriting module callback values with nil in hybrid mode defp maybe_put(map, _key, nil), do: map defp maybe_put(map, key, value), do: Map.put(map, key, value) defp build_module_content(context, channel, module, _event_config, recipient) do base_content = %{ transport: channel.transport, audience: channel.audience } # Add transport-specific content from callback module transport_content = case channel.transport do :email -> # Get variant for template resolution # Prefer channel.variant (from inline DSL) over EventResolver callback variant = channel.variant || EventResolver.template_variant(module, context, channel) # Locale priority: channel.locale > channel.locale_from (dynamic) > # recipient.locale > context.locale locale = resolve_channel_locale(channel, context, recipient) # Prepare template assigns using EventResolver base_assigns = safe_prepare_template_assigns(module, context, channel) # Merge context data and variables into assigns (variables take precedence) assigns = base_assigns |> Map.merge(Context.template_assigns(context)) # Get subject early and add to assigns so layout template can access it subject = module.subject(context, channel) assigns_with_subject = Map.put(assigns, :subject, subject) # Check if module has body_html/body_text callbacks (for test modules without templates) {html_body, text_body} = if EventResolver.has_body_callbacks?(module) do # Use callbacks directly (for test modules or simple modules) {EventResolver.body_html(module, context, channel), EventResolver.body_text(module, context, channel)} else # Try to render templates using TemplateResolver (uses compiled templates or event_dir) # For hybrid mode with inline DSL, templates may not be found - that's OK, # inline DSL rendering will handle it # Derive OTP app from event_id for template resolution otp_app = derive_otp_app_from_event_id(context.event_id) # Derive event directory from module name for file-based loading in development event_dir = derive_event_dir_from_module(module, otp_app) html = case AshDispatch.TemplateResolver.render( event_module: module, event_dir: event_dir, otp_app: otp_app, format: :html, transport: :email, variant: variant, locale: locale, assigns: assigns_with_subject ) do {:ok, rendered} -> rendered {:error, :template_not_found} -> nil _ -> nil end text = case AshDispatch.TemplateResolver.render( event_module: module, event_dir: event_dir, otp_app: otp_app, format: :text, transport: :email, variant: variant, locale: locale, assigns: assigns_with_subject ) do {:ok, rendered} -> rendered {:error, :template_not_found} -> nil _ -> nil end {html, text} end # Get from as tuple and convert to map for JSON encoding {from_name, from_email} = module.from(context, channel) %{ subject: subject, from: %{"name" => from_name, "email" => from_email}, html_body: html_body, text_body: text_body } :in_app -> %{ title: module.notification_title(context, channel), message: module.notification_message(context, channel), action_url: module.action_url(context, channel), action_label: module.action_label(context, channel), notification_type: get_notification_type(module, context) } _ -> # For other transports, use basic message %{ message: module.notification_message(context, channel) } end Map.merge(base_content, transport_content) end defp get_notification_type(module, context) do # Use EventResolver for safe callback execution with default :info EventResolver.notification_type(module, context) end # F1 — transport dispatch routes through `AshDispatch.Transport.Registry` # (compile-time `atom => module` map). Adding a transport is one new # file + one entry in the registry's @transports list — no edit here. defp dispatch_to_transport(receipt, context, channel, event_config) do case AshDispatch.Transport.Registry.module_for(channel.transport) do {:ok, module} -> module.deliver(receipt, context, channel, event_config) :error -> Logger.warning("Unknown transport: #{channel.transport}, skipping delivery") receipt |> Ash.Changeset.for_update(:skip, %{ error_message: "Unknown transport: #{channel.transport}" }) |> Ash.update() end end # Helper to extract user from data map using Ash introspection # No hardcoded patterns - derives from Ash resource relationships defp extract_user_from_data(data) do user_module = Config.user_module() if is_nil(user_module) do Logger.warning("No :user_module configured in :ash_dispatch config") nil else # Strategy 1: Check if any value in data IS the user module struct Enum.find_value(data, fn {_key, value} -> if is_struct(value) && value.__struct__ == user_module do value end end) || # Strategy 2: Use Ash introspection to find user via relationships find_user_via_ash_relationships(data, user_module) || # Strategy 3: Accept bare map with :id under :user or :actor key # (manual pipeline events pass %{user: %{id: user_id}} without loading the full struct) extract_bare_user_map(data) end end defp extract_bare_user_map(data) do case data do %{user: %{id: _} = bare} when not is_struct(bare) -> bare %{actor: %{id: _} = bare} when not is_struct(bare) -> bare _ -> nil end end # Find user by introspecting Ash resource relationships # Works for ANY resource that has a relationship to User module defp find_user_via_ash_relationships(data, user_module) do Enum.find_value(data, fn {_key, resource} -> # Only process Ash resources if is_struct(resource) && Ash.Resource.Info.resource?(resource.__struct__) do # Get all relationships defined on this resource relationships = Ash.Resource.Info.relationships(resource.__struct__) # Find any relationship pointing to the configured User module user_relationship = Enum.find(relationships, fn rel -> rel.destination == user_module end) # Extract user from that relationship if found if user_relationship do Map.get(resource, user_relationship.name) end end end) end # Counter broadcasting integration defp broadcast_counters(context, channel, event_config) do counter_broadcaster = Config.counter_broadcaster() if counter_broadcaster && Code.ensure_loaded?(counter_broadcaster) && function_exported?(counter_broadcaster, :broadcast, 3) do # Resolve module with runtime fallback (handles compilation order issues) module = resolve_event_module(event_config, context) # Hybrid mode: prefer inline DSL counters over module callback counters = cond do # Check if channel has counters defined in inline DSL is_list(channel.counters) and channel.counters != [] -> channel.counters # Fall back to event module callback using EventResolver not is_nil(module) -> EventResolver.counters(module, context, channel) # No counters defined true -> [] end # Broadcast each counter via configured broadcaster Enum.each(counters, fn counter_name -> try do counter_broadcaster.broadcast(counter_name, context, channel) rescue error -> Logger.warning(""" Failed to broadcast counter Counter: #{counter_name} Event: #{context.event_id} Error: #{inspect(error)} """) end end) end :ok end # Apply channel-level load to the primary resource in context defp apply_channel_load(context, %{load: []} = _channel), do: context defp apply_channel_load(context, %{load: nil} = _channel), do: context defp apply_channel_load(context, %{load: load} = _channel) when is_list(load) do # Get the primary resource from context.data using resource_key resource_key = context.resource_key case Map.get(context.data, resource_key) do nil -> # No primary resource to load, return context as-is context record when is_struct(record) -> # Load additional relationships on the record # Use Ash.Resource.Info.domain/1 to safely get the domain case Ash.Resource.Info.domain(record.__struct__) do nil -> Logger.warning(""" Cannot load channel-level relationships #{inspect(load)} Resource #{inspect(record.__struct__)} has no domain configured Continuing with unloaded record... """) context domain -> case Ash.load(record, load, domain: domain, authorize?: false) do {:ok, loaded_record} -> # Update context with loaded record updated_data = Map.put(context.data, resource_key, loaded_record) %{context | data: updated_data} {:error, error} -> Logger.warning(""" Failed to load channel-level relationships #{inspect(load)} Error: #{inspect(error)} Continuing with unloaded record... """) context end end _other -> # Not a struct, can't load context end end defp apply_channel_load(context, _channel), do: context # Always resolve event module at runtime via EventResolver # Compile-time resolution is unreliable due to module compilation order # (event modules compile after resources, so Code.ensure_loaded? fails) # Warning is only logged in dispatch_event.ex on first resolution failure defp resolve_event_module(_event_config, context) do case EventResolver.find_module(context.event_id) do {:ok, m} -> m {:error, :not_found} -> nil end end # Resolve locale for a channel with priority: # 1. channel.locale - static locale configured on channel # 2. channel.locale_from - dynamic locale from record field # 3. recipient.locale - per-recipient preference (NEW in 0.4.5; only # consulted when the recipient struct carries a non-nil `:locale` # field — typically a User record for `audience: :user`) # 4. context.locale - event-level locale (already resolved from record # or default) @doc false def resolve_channel_locale(channel, context, recipient) do cond do # Static locale on channel has highest priority channel.locale -> channel.locale # Dynamic locale from record field. When the field is set but # the record's value is nil/missing, cascade through the rest of # the chain (recipient.locale → context.locale) instead of # bailing on `context.locale`. Makes `locale_from` describe a # *preference*, not a hard constraint. channel.locale_from -> record_locale(context, channel.locale_from) || recipient_locale(recipient) || context.locale # Per-recipient locale (e.g. User.locale) — wins over the # event/resource-level fallback so each recipient sees their own # language. locale = recipient_locale(recipient) -> locale # Fall back to context locale (event-level) true -> context.locale end end # Extract `:locale` from a recipient struct/map. Returns nil if the # field is missing, nil, or not a non-empty binary — we explicitly # reject empty strings so a blank `User.locale` doesn't shadow # downstream fallbacks. defp recipient_locale(nil), do: nil defp recipient_locale(recipient) when is_map(recipient) do case Map.get(recipient, :locale) do locale when is_binary(locale) and locale != "" -> locale _ -> nil end end defp recipient_locale(_), do: nil # Read `locale_field` off `context.data[resource_key]`. Returns nil # (not `context.locale`) when missing so the caller controls the # cascade. Treats `%Ash.NotLoaded{}` as nil via RecordReader. defp record_locale(context, locale_field) do resource_key = context.resource_key case Map.get(context.data, resource_key) do record when is_map(record) -> RecordReader.safe_get(record, locale_field) _ -> nil end end # Wrap prepare_template_assigns with helpful error messages for unloaded relationships # Uses EventResolver for safe callback execution, but adds extra error handling for NotLoaded defp safe_prepare_template_assigns(module, context, channel) do # First check if the module exports the function if EventResolver.exports?(module, :prepare_template_assigns, 2) do try do module.prepare_template_assigns(context, channel) rescue e in KeyError -> reraise_with_load_hint(e, context, __STACKTRACE__) e in UndefinedFunctionError -> # Happens when trying to call functions on NotLoaded struct if String.contains?(Exception.message(e), "Ash.NotLoaded") do reraise_with_load_hint(e, context, __STACKTRACE__) else reraise e, __STACKTRACE__ end e in Protocol.UndefinedError -> # Happens when trying to enumerate NotLoaded if String.contains?(Exception.message(e), "Ash.NotLoaded") do reraise_with_load_hint(e, context, __STACKTRACE__) else reraise e, __STACKTRACE__ end end else %{} end end defp reraise_with_load_hint(original_error, context, stacktrace) do event_id = context.event_id original_message = Exception.message(original_error) message = """ Failed to prepare template assigns for event "#{event_id}". #{original_message} This usually means a relationship wasn't loaded. Add it to the `load:` option in your event DSL: dispatch do event :your_event, trigger_on: :your_action, load: [:user, items: :product], # <-- Add missing relationships here ... end Or for nested relationships: load: [:user, product_order_items: :product] """ reraise RuntimeError.exception(message), stacktrace end # Reverse-lookup: module → event_id via EventRegistry. # Used by the module-typed dispatch/3 overload to resolve dispatch IDs # from event modules instead of hand-typed strings. @spec resolve_event_id_for_module(module()) :: {:ok, String.t()} | :error defp resolve_event_id_for_module(module) do case Enum.find(AshDispatch.EventRegistry.get_event_modules(), fn {_id, mod} -> mod == module end) do {event_id, _module} when is_binary(event_id) -> {:ok, event_id} _ -> :error end end end