ash_dispatch

defmodule AshDispatch.Transports.InApp do
  use AshDispatch.Transport, atom: :in_app, skip_receipt?: false

  @moduledoc """
  In-app notification transport.

  Creates in-app notifications immediately (synchronous).

  ## Behavior

  1. Checks user preferences (if event is user_configurable)
  2. Resolves recipients from context
  3. Creates Notification records
  4. Updates receipt status to `:sent`

  ## Status Flow

  ```
  pending → sent (success)
          ↘ failed (error)
          ↘ skipped (user opted out)
  ```

  ## Example

      receipt = %{
        content: %{
          title: "Order Created",
          message: "Your order #1234 is being processed",
          action_url: "/orders/1234",
          notification_type: :success
        }
      }

      InApp.deliver(receipt, context, channel, event_config)
      # -> Creates Notification records
      # -> Returns {:ok, updated_receipt}
  """

  import AshDispatch.ContentMap

  alias AshDispatch.Config

  require Logger

  @doc """
  Delivers an in-app notification.

  ## Parameters

  - `receipt` - DeliveryReceipt map
  - `context` - Event context
  - `channel` - Channel configuration
  - `event_config` - Event configuration

  ## Returns

  - `{:ok, updated_receipt}` on success
  - `{:error, reason}` on failure
  """
  def deliver(receipt, context, channel, event_config) do
    # Check user preferences first
    if not AshDispatch.UserPreference.allows?(context, channel, event_config) do
      Logger.info("User opted out of #{context.event_id} via #{channel.transport}, skipping")

      updated_receipt =
        receipt
        |> Ash.Changeset.for_update(:skip, %{error_message: "user_opted_out"})
        |> Ash.update!()

      {:ok, updated_receipt}
    else
      # Receipt now corresponds to a single recipient (user_id in receipt)
      # Create one notification for this recipient
      invalidates = event_config[:invalidates] || []
      result = create_notification_for_receipt(receipt, context, channel, invalidates)

      # Update receipt status and link notification_id
      updated_receipt = update_receipt_with_notification(receipt, result)

      {:ok, updated_receipt}
    end
  rescue
    error ->
      Logger.error("""
      InApp transport failed
      Event: #{context.event_id}
      Error: #{inspect(error)}
      """)

      {:error, error}
  end

  @doc """
  Retry a failed in-app delivery directly from a stored receipt.

  In-app delivery is synchronous (DB write + PubSub broadcast), so we
  re-attempt the Notification.create directly rather than going through Oban.

  Uses the receipt's stored content and idempotency_key to prevent duplicates.

  Returns `:ok` on success or `{:error, reason}` on failure.
  """
  def retry_from_receipt(receipt) do
    user_id = receipt.user_id
    content = receipt.content || %{}

    if is_nil(user_id) do
      {:error, :no_user_id}
    else
      # Rebuild idempotency key from receipt fields to match original delivery format.
      # Original format: "event_id:source_id:audience:user_id" or "event_id:audience:user_id"
      idempotency_key =
        case receipt do
          %{source_id: source_id} when is_binary(source_id) and source_id != "" ->
            "#{receipt.event_id}:#{source_id}:#{receipt.audience}:#{user_id}"

          _ ->
            "#{receipt.event_id}:#{receipt.audience}:#{user_id}"
        end

      notification_attrs = %{
        user_id: user_id,
        title: get_content(content, :title),
        message: get_content(content, :message),
        action_url: get_content(content, :action_url),
        action_label: get_content(content, :action_label),
        event_id: receipt.event_id,
        source: receipt.event_id,
        type: get_notification_type(content),
        metadata: get_content(content, :metadata) || %{},
        idempotency_key: idempotency_key
      }

      notification_resource = Config.notification_resource()

      case notification_resource
           |> Ash.Changeset.for_create(:create, notification_attrs)
           |> Ash.create() do
        {:ok, notification} ->
          # Broadcast and mark receipt as sent
          invalidates = Map.get(content, :invalidates, [])
          broadcast_notification(notification, invalidates)

          receipt
          |> Ash.Changeset.for_update(:mark_sent, %{notification_id: notification.id})
          |> Ash.update(authorize?: false)

          :ok

        {:error, error} ->
          # Duplicate idempotency key means the notification was already delivered —
          # treat as success and mark the receipt as sent.
          if idempotency_conflict?(error) do
            Logger.debug(
              "InApp retry: notification already exists (idempotency key match), marking receipt as sent: receipt_id=#{receipt.id}"
            )

            receipt
            |> Ash.Changeset.for_update(:mark_sent, %{})
            |> Ash.update(authorize?: false)

            :ok
          else
            Logger.error(
              "InApp retry failed: receipt_id=#{receipt.id}, reason=#{inspect(error)}"
            )

            {:error, error}
          end
      end
    end
  end

  # Private functions

  # Create notification for the receipt (one receipt = one recipient now)
  defp create_notification_for_receipt(receipt, context, channel, invalidates) do
    # Receipt now has the user_id of the recipient
    user_id = receipt.user_id

    # Skip in-app notifications if no user_id (external recipients, webhooks, etc.)
    cond do
      is_nil(user_id) ->
        Logger.warning("""
        Skipping in-app notification creation: no user_id
        Event: #{context.event_id}
        Recipient: #{receipt.recipient}

        In-app notifications require a valid user_id. This receipt will be skipped.
        """)

        # Skip the receipt since in-app notifications require user_id
        {:error, :no_user_id}

      true ->
        # Generate idempotency key to prevent duplicates when user receives
        # notifications from multiple audiences (e.g., user who is also admin)
        # Format: "event_id:resource_id:audience:user_id" or "event_id:audience:user_id" if no resource_id
        idempotency_key =
          case extract_resource_id(context) do
            nil -> "#{context.event_id}:#{channel.audience}:#{user_id}"
            resource_id -> "#{context.event_id}:#{resource_id}:#{channel.audience}:#{user_id}"
          end

        # Build metadata from event config + context priority
        metadata =
          (receipt.content[:metadata] || %{})
          |> Map.put(:priority, context.priority || :standard)

        notification_attrs = %{
          user_id: user_id,
          title: get_content(receipt.content, :title),
          message: get_content(receipt.content, :message),
          action_url: get_content(receipt.content, :action_url),
          action_label: get_content(receipt.content, :action_label),
          event_id: context.event_id,
          source: context.event_id,
          type: get_notification_type(receipt.content),
          metadata: metadata,
          idempotency_key: idempotency_key
        }

        # Create Notification record via Ash
        notification_resource = Config.notification_resource()

        case notification_resource
             |> Ash.Changeset.for_create(:create, notification_attrs)
             |> Ash.create() do
          {:ok, notification} ->
            Logger.debug("""
            Created in-app notification:
            User: #{notification.user_id}
            Title: #{notification.title}
            Message: #{notification.message}
            """)

            # Broadcast to user's channel with invalidation keys
            broadcast_notification(notification, invalidates)

            {:ok, notification}

          {:error, error} ->
            # Idempotency conflict means the notification already exists —
            # treat as success (race condition or duplicate dispatch).
            if idempotency_conflict?(error) do
              Logger.debug(
                "InApp: notification already exists (idempotency key match), treating as success: event=#{context.event_id}, user=#{user_id}"
              )

              # Look up existing notification to link to receipt
              case find_by_idempotency_key(notification_resource, idempotency_key) do
                {:ok, existing} -> {:ok, existing}
                _ -> {:ok, :already_exists}
              end
            else
              Logger.error("""
              Failed to create in-app notification:
              User: #{notification_attrs.user_id}
              Error: #{inspect(error)}
              """)

              {:error, error}
            end
        end
    end
  end

  # Update receipt with notification_id and mark as sent
  defp update_receipt_with_notification(receipt, result) do
    case result do
      {:ok, %{id: notification_id}} ->
        receipt
        |> Ash.Changeset.for_update(:mark_sent, %{notification_id: notification_id})
        |> Ash.update!()

      {:ok, :already_exists} ->
        # Idempotency conflict — notification exists but we couldn't look it up.
        # Mark as sent without linking notification_id.
        receipt
        |> Ash.Changeset.for_update(:mark_sent, %{})
        |> Ash.update!()

      {:error, :no_user_id} ->
        # Skip receipts for external recipients without user_ids
        receipt
        |> Ash.Changeset.for_update(:skip, %{
          error_message: "In-app notifications require user_id (external recipient)"
        })
        |> Ash.update!()

      {:error, reason} ->
        receipt
        |> Ash.Changeset.for_update(:mark_failed, %{error_message: inspect(reason)})
        |> Ash.update!()
    end
  end

  # Extract the primary resource ID from context data
  # Used for idempotency keys to prevent duplicate notifications
  defp extract_resource_id(%{data: data}) when is_map(data) do
    # Find the first value in data map that has an :id field
    data
    |> Map.values()
    |> Enum.find_value(fn
      %{id: id} when is_binary(id) -> id
      _ -> nil
    end)
  end

  defp extract_resource_id(_), do: nil

  # Get notification type, converting string values to atoms and defaulting to :info
  defp get_notification_type(content) do
    case get_content(content, :notification_type) do
      type when is_atom(type) -> type
      "success" -> :success
      "info" -> :info
      "warning" -> :warning
      "error" -> :error
      _ -> :info
    end
  end

  # Broadcast notification to user's channel in JSON-serializable format
  defp broadcast_notification(notification, invalidates) do
    pubsub_module = Config.pubsub_module()

    if pubsub_module do
      serialized = %{
        id: notification.id,
        type: notification.type,
        title: notification.title,
        message: notification.message,
        read: notification.read,
        source: notification.source,
        occurredAt: notification.occurred_at,
        insertedAt: notification.inserted_at,
        metadata: notification.metadata || %{},
        actionLabel: notification.action_label,
        actionUrl: notification.action_url,
        invalidates: invalidates
      }

      topic = "#{Config.channel_topic()}:#{notification.user_id}"

      pubsub_module.broadcast(
        topic,
        "new_notification",
        serialized
      )
    end
  end

  # Look up an existing notification by idempotency key
  defp find_by_idempotency_key(notification_resource, key) do
    require Ash.Query

    notification_resource
    |> Ash.Query.filter(idempotency_key == ^key)
    |> Ash.Query.limit(1)
    |> Ash.read(authorize?: false)
    |> case do
      {:ok, [notification]} -> {:ok, notification}
      _ -> {:error, :not_found}
    end
  end

  # Check if an Ash error contains a unique constraint violation on idempotency_key.
  # Handles both direct error structs and nested Splode error wrappers.
  defp idempotency_conflict?(error) do
    errors = extract_errors(error)

    Enum.any?(errors, fn
      %{field: :idempotency_key, message: "has already been taken"} ->
        true

      %{private_vars: vars} when is_list(vars) ->
        constraint = Keyword.get(vars, :constraint, "")
        String.contains?(to_string(constraint), "idempotency")

      error_item ->
        # Fallback: check string representation for idempotency constraint violations
        error_str = inspect(error_item)

        String.contains?(error_str, "idempotency_key") and
          String.contains?(error_str, "has already been taken")
    end)
  end

  # Extract flat list of errors from potentially nested Ash/Splode error structures
  defp extract_errors(%{errors: errors}) when is_list(errors) do
    Enum.flat_map(errors, fn
      %{errors: nested} when is_list(nested) -> extract_errors(%{errors: nested})
      error -> [error]
    end)
  end

  defp extract_errors(error), do: [error]
end