phoenix_kit

defmodule PhoenixKit.Integrations do
  @moduledoc """
  Centralized management of external service integrations.

  Stores credentials (OAuth tokens, API keys, bot tokens, etc.) using the
  existing `PhoenixKit.Settings` system with `value_json` JSONB storage.
  Each integration row's `key` column is just its UUID; the `module`
  column is stamped `"integrations"` and the JSONB body holds
  `{"provider", "name", "auth_type", "status", ...}`. There is no
  composite key shape — provider and name are pure JSONB.

  Connections are referenced by the storage row's UUID. Names are pure
  user-chosen labels with no system semantics — any string is allowed
  (including spaces and duplicates within a provider); consumer modules
  pin to UUIDs that survive renames.

  ## Auth types supported

  - `:oauth2` — Google, Microsoft, Slack, etc. (client_id/secret + access/refresh tokens)
  - `:api_key` — OpenRouter, Stripe, SendGrid, etc. (single API key)
  - `:key_secret` — AWS, Twilio, etc. (access key + secret key)
  - `:bot_token` — Telegram, Discord, etc. (single bot token)
  - `:credentials` — SMTP, databases, etc. (freeform credential map)

  ## Usage

  Consumer modules (AI endpoints, document creator, etc.) store an
  integration's UUID on their own records and resolve credentials by UUID:

      # Look up the row by uuid (the stable reference consumers store)
      {:ok, %{provider: "openrouter", name: "default", data: data}} =
        PhoenixKit.Integrations.get_integration_by_uuid(integration_uuid)

      # Get credentials for API calls — accepts either a uuid or a
      # `provider:name` shape
      {:ok, creds} = PhoenixKit.Integrations.get_credentials(integration_uuid)
      # => %{"access_token" => "ya29...", "token_type" => "Bearer", ...}

      # Make an authenticated request with auto-refresh on 401
      {:ok, response} =
        PhoenixKit.Integrations.authenticated_request(integration_uuid, :get, url)

  ## Renaming and removing

  Any connection can be renamed or removed — there's no privileged
  `"default"` name. The storage row's UUID stays stable across renames,
  so consumer references don't break:

      {:ok, _} = PhoenixKit.Integrations.rename_connection(uuid, "work")
      :ok = PhoenixKit.Integrations.remove_connection(uuid)

  ## API shape (uuid-strict)

  Every operation past row creation takes the row's `uuid`. The only
  exceptions are:

  - `add_connection/3` — row birth, no uuid exists yet
  - `get_integration/1`, `find_uuid_by_provider_name/1` — read shims for
    legacy `migrate_legacy/0` callbacks that walk pre-uuid data shapes.
    Since names are no longer unique, these now return first-match.

  The `key` column is the row's UUIDv7 — collisions are structurally
  impossible. JSONB `provider`/`name` are pure data; corruption there
  affects only this row's content, never routing.
  """

  use Gettext, backend: PhoenixKitWeb.Gettext

  import Ecto.Query, only: [from: 2]

  require Logger

  alias PhoenixKit.Integrations.Encryption
  alias PhoenixKit.Integrations.Events
  alias PhoenixKit.Integrations.OAuth
  alias PhoenixKit.Integrations.Providers
  alias PhoenixKit.Settings
  alias PhoenixKit.Settings.Queries
  alias PhoenixKit.Settings.Setting

  @settings_module "integrations"
  @http_timeout 15_000

  # ---------------------------------------------------------------------------
  # Reading credentials
  # ---------------------------------------------------------------------------

  @doc """
  Get the full integration data for a provider.

  Returns the entire JSON blob including credentials, status, and metadata.
  Misses return `:not_configured` (or `:deleted` for uuid input) — there's
  no on-read legacy-shape migration in core anymore. Modules with legacy
  data own their own migration via the `migrate_legacy/0` callback on
  `PhoenixKit.Module` (orchestrated by
  `PhoenixKit.ModuleRegistry.run_all_legacy_migrations/0`).
  """
  @spec get_integration(String.t()) ::
          {:ok, map()} | {:error, :not_configured | :invalid_provider_key}
  def get_integration(provider_key) when is_binary(provider_key) and provider_key != "" do
    if uuid?(provider_key) do
      case Settings.get_json_setting_by_uuid(provider_key) do
        %{} = data -> {:ok, Encryption.decrypt_fields(data)}
        _ -> {:error, :not_configured}
      end
    else
      # `"provider:name"` string lookup — first-match by name. Names
      # are no longer unique, so this returns the first row whose JSONB
      # name matches. Used by legacy `migrate_legacy/0` callbacks; new
      # callers should use uuid-based lookups.
      case find_uuid_by_provider_name(provider_key) do
        {:ok, uuid} ->
          case Settings.get_json_setting_by_uuid(uuid) do
            %{} = data -> {:ok, Encryption.decrypt_fields(data)}
            _ -> {:error, :not_configured}
          end

        _ ->
          {:error, :not_configured}
      end
    end
  end

  def get_integration(_), do: {:error, :invalid_provider_key}

  @doc """
  Look up an integration row by its settings UUID and return a normalized
  shape with `provider`, `name`, `data`, and the original `uuid`.

  Used by the integration form LV (route `/admin/settings/integrations/:uuid`)
  so the URL is stable across renames — the human-readable `name` lives in
  the JSONB blob, the URL stays pinned to the row's storage UUID.
  """
  @spec get_integration_by_uuid(String.t()) ::
          {:ok, %{uuid: String.t(), provider: String.t(), name: String.t(), data: map()}}
          | {:error, :not_configured | :invalid_uuid}
  def get_integration_by_uuid(uuid) when is_binary(uuid) and uuid != "" do
    # The storage row is identified solely by its uuid. Provider and name
    # both live in JSONB (`data["provider"]` / `data["name"]`). The row's
    # `module` column is the discriminator that distinguishes integration
    # rows from other settings.
    case Queries.get_setting_by_uuid(uuid) do
      %{module: "integrations", value_json: data} when is_map(data) ->
        decrypted = Encryption.decrypt_fields(data)
        provider = Map.get(decrypted, "provider", "")
        name = Map.get(decrypted, "name", "")

        {:ok, %{uuid: uuid, provider: provider, name: name, data: decrypted}}

      _ ->
        {:error, :not_configured}
    end
  end

  def get_integration_by_uuid(_), do: {:error, :invalid_uuid}

  @doc """
  Resolve a `provider:name`-style reference to the storage row's uuid.

  Used by consumer modules' `migrate_legacy/0` implementations to walk
  legacy name-string references and rewrite them to uuid references.
  Accepts a few input shapes for convenience:

  - `"openrouter:work"` — full provider:name pair
  - `"openrouter"` — bare provider, treated as `provider:default`
  - `{"openrouter", "work"}` — explicit tuple

  Returns `{:ok, uuid}` if a matching row exists, `{:error, :not_found}`
  if not, `{:error, :invalid}` for malformed input. Does NOT auto-pick
  an arbitrary connection when multiple match — that's not the
  caller's intent here.
  """
  @spec find_uuid_by_provider_name(String.t() | {String.t(), String.t()}) ::
          {:ok, String.t()} | {:error, :not_found | :invalid}
  def find_uuid_by_provider_name(input)

  def find_uuid_by_provider_name({provider, name})
      when is_binary(provider) and is_binary(name) and provider != "" and name != "" do
    case Enum.find(list_connections(provider), &(&1.name == name)) do
      %{uuid: uuid} -> {:ok, uuid}
      _ -> {:error, :not_found}
    end
  end

  def find_uuid_by_provider_name(string) when is_binary(string) and string != "" do
    case String.split(string, ":", parts: 2) do
      [provider, name] when provider != "" and name != "" ->
        find_uuid_by_provider_name({provider, name})

      [provider] when provider != "" ->
        # Bare provider — first connection of that provider (sorted by
        # name, lowercase). Used as the "any connection of this kind"
        # fallback by legacy paths.
        case list_connections(provider) do
          [%{uuid: uuid} | _] -> {:ok, uuid}
          _ -> {:error, :not_found}
        end

      _ ->
        {:error, :invalid}
    end
  end

  def find_uuid_by_provider_name(_), do: {:error, :invalid}

  @doc """
  Resolves a binary that may be EITHER an integration row's uuid OR a
  `provider:name` string into the canonical row uuid.

  This is the dual-input lookup that consumer modules' lazy-promotion
  paths and migration sweeps converge on — code that reads a legacy
  string from a column where the operator might have stuffed a uuid
  pre-V107, or a `provider:name` shape pre-uuid-strict, or a bare
  provider key. Each consumer used to copy the same regex + dispatch
  pair into its own helper; this primitive centralises it so a future
  provider doesn't tempt a third copy.

  Returns `{:ok, uuid}` if the input resolves to a current row,
  `{:error, :not_found}` if it parses cleanly but no matching row
  exists, `{:error, :invalid}` for malformed input (empty string, nil,
  non-binary).

  ## Examples

      iex> resolve_to_uuid("019b669c-3c9d-7256-8ed1-edbc6ae29703")
      {:ok, "019b669c-3c9d-7256-8ed1-edbc6ae29703"}  # already-uuid path

      iex> resolve_to_uuid("openrouter:default")
      {:ok, "..."}  # provider:name path → find_uuid_by_provider_name

      iex> resolve_to_uuid("openrouter")
      {:ok, "..."}  # bare provider, treated as provider:default

  See `find_uuid_by_provider_name/1` for the provider:name half of the
  lookup. The split exists because that primitive doesn't handle the
  "input is already a uuid" case — it'd treat `"019b669c-..."` as a
  provider name and search `integration:019b669c-...:default`.
  """
  @spec resolve_to_uuid(String.t()) ::
          {:ok, String.t()} | {:error, :not_found | :invalid}
  def resolve_to_uuid(input) when is_binary(input) and input != "" do
    if uuid?(input) do
      case get_integration_by_uuid(input) do
        {:ok, _} -> {:ok, input}
        _ -> {:error, :not_found}
      end
    else
      find_uuid_by_provider_name(input)
    end
  end

  def resolve_to_uuid(_), do: {:error, :invalid}

  @doc """
  Get credentials for a provider, suitable for making API calls.

  Returns the full integration data map. The caller extracts what it needs
  based on the auth type (e.g., `"access_token"` for OAuth, `"api_key"` for API key).
  """
  @spec get_credentials(String.t()) :: {:ok, map()} | {:error, :not_configured | :deleted}
  def get_credentials(provider_key) when is_binary(provider_key) and provider_key != "" do
    is_uuid = uuid?(provider_key)

    data =
      if is_uuid do
        Settings.get_json_setting_by_uuid(provider_key)
      else
        case find_uuid_by_provider_name(provider_key) do
          {:ok, uuid} -> Settings.get_json_setting_by_uuid(uuid)
          _ -> nil
        end
      end

    case data do
      %{} = data when map_size(data) > 0 ->
        decrypted = Encryption.decrypt_fields(data)
        if has_credentials?(decrypted), do: {:ok, decrypted}, else: {:error, :not_configured}

      _ ->
        # uuid lookup that missed → row was deleted; bare-provider /
        # provider:name lookup that missed → never configured. Both
        # surface as the appropriate atom so callers can distinguish.
        if is_uuid, do: {:error, :deleted}, else: {:error, :not_configured}
    end
  end

  def get_credentials(_), do: {:error, :not_configured}

  @doc """
  Check if an integration is connected and has valid credentials.
  """
  @spec connected?(String.t()) :: boolean()
  def connected?(provider_key) when is_binary(provider_key) do
    case get_credentials(provider_key) do
      {:ok, _} -> true
      _ -> false
    end
  end

  def connected?(_), do: false

  # ---------------------------------------------------------------------------
  # Setup (saving app-level credentials)
  # ---------------------------------------------------------------------------

  @doc """
  Save setup credentials for an existing connection (referenced by uuid).

  For OAuth providers, this saves client_id/client_secret.
  For API key providers, this saves the api_key.
  For bot token providers, this saves the bot_token.

  Merges with existing data to preserve any previously obtained tokens.
  Sets status to "disconnected" if no runtime credentials exist yet.

  The connection must exist (`add_connection/3` is the row-birth path).
  Returns `{:error, :not_configured}` if the uuid doesn't resolve.
  """
  @spec save_setup(String.t(), map(), String.t() | nil) ::
          {:ok, map()} | {:error, :not_configured | :invalid_uuid | term()}
  def save_setup(uuid, attrs, actor_uuid \\ nil) when is_binary(uuid) and is_map(attrs) do
    with {:ok, %{provider: base_provider, name: name, setting: setting, data: existing}} <-
           resolve_uuid(uuid) do
      provider = Providers.get(base_provider)

      data =
        existing
        |> Map.merge(attrs)
        |> Map.put("provider", base_provider)
        |> Map.put("name", name)
        |> Map.put("auth_type", provider && Atom.to_string(provider.auth_type))
        |> maybe_set_status(provider)
        |> maybe_set_connected_at()

      case save_integration(setting, data) do
        {:ok, saved} = result ->
          Events.broadcast_setup_saved(base_provider, saved)

          log_activity(
            "integration.setup_saved",
            base_provider,
            name,
            %{"status" => saved["status"]},
            "manual",
            actor_uuid
          )

          result

        error ->
          error
      end
    end
  end

  # ---------------------------------------------------------------------------
  # OAuth flow
  # ---------------------------------------------------------------------------

  @doc """
  Build the OAuth authorization URL for a connection (by uuid).

  Accepts an optional `state` parameter for CSRF protection. Use
  `PhoenixKit.Integrations.OAuth.generate_state/0` to generate one,
  store it in the session or socket assigns, and verify it when the
  callback arrives.
  """
  @spec authorization_url(String.t(), String.t(), String.t() | nil, String.t() | nil) ::
          {:ok, String.t()} | {:error, term()}
  def authorization_url(uuid, redirect_uri, extra_scopes \\ nil, state \\ nil)
      when is_binary(uuid) do
    with {:ok, %{provider: base_provider, data: data}} <- resolve_uuid(uuid),
         {:ok, provider} <- fetch_provider(base_provider) do
      OAuth.authorization_url(provider.oauth_config, data, redirect_uri, extra_scopes, state)
    end
  end

  @doc """
  Exchange an OAuth authorization code for tokens and save them on the
  connection identified by `uuid`.
  """
  @spec exchange_code(String.t(), String.t(), String.t(), String.t() | nil) ::
          {:ok, map()} | {:error, term()}
  def exchange_code(uuid, code, redirect_uri, actor_uuid \\ nil) when is_binary(uuid) do
    with {:ok, %{provider: base_provider, name: name, setting: setting, data: data}} <-
           resolve_uuid(uuid),
         {:ok, provider} <- fetch_provider(base_provider),
         {:ok, token_data} <- OAuth.exchange_code(provider.oauth_config, data, code, redirect_uri) do
      userinfo = fetch_userinfo_safe(provider, token_data["access_token"])

      updated =
        data
        |> Map.merge(token_data)
        |> Map.put("status", "connected")
        |> Map.put("connected_at", DateTime.utc_now() |> DateTime.to_iso8601())
        |> Map.put(
          "scopes",
          provider.oauth_config[:default_scopes] || provider.oauth_config["default_scopes"]
        )
        |> maybe_set_userinfo(userinfo)

      case save_integration(setting, updated) do
        {:ok, saved} = result ->
          Events.broadcast_connected(base_provider, saved)

          log_activity(
            "integration.connected",
            base_provider,
            name,
            %{"account" => saved["external_account_id"]},
            "manual",
            actor_uuid
          )

          result

        error ->
          error
      end
    end
  end

  @doc """
  Refresh an expired OAuth access token and save the new one.

  On failure, stamps the integration record with `status: "error"` and a
  human-readable `validation_status` so the UI reflects the broken state
  without waiting for an admin to click "Test Connection".
  On success following a previously-errored state, auto-recovers the status
  back to `"connected"`.
  """
  @spec refresh_access_token(String.t()) :: {:ok, String.t()} | {:error, term()}
  def refresh_access_token(uuid) when is_binary(uuid) do
    result =
      with {:ok, %{provider: base_provider, name: name, setting: setting, data: data}} <-
             resolve_uuid(uuid),
           {:ok, provider} <- fetch_provider(base_provider),
           {:ok, new_token, updated_fields} <-
             OAuth.refresh_access_token(provider.oauth_config, data) do
        updated = Map.merge(data, updated_fields)
        save_integration(setting, updated)

        log_activity("integration.token_refreshed", base_provider, name, %{}, "auto", nil)

        {:ok, new_token}
      end

    case result do
      {:ok, _token} ->
        maybe_record_recovery(uuid)
        result

      {:error, reason} ->
        record_refresh_failure(uuid, reason)
        result
    end
  end

  @doc """
  Disconnect a connection (remove tokens, keep setup credentials).

  For OAuth: removes access_token, refresh_token, keeps client_id/client_secret.
  For API key/bot token: removes the key entirely.

  No-op when the uuid doesn't resolve (already gone).
  """
  @spec disconnect(String.t(), String.t() | nil) :: :ok
  def disconnect(uuid, actor_uuid \\ nil) when is_binary(uuid) do
    case resolve_uuid(uuid) do
      {:ok, %{provider: base_provider, name: name, setting: setting, data: data}} ->
        auth_type = data["auth_type"]

        cleaned =
          case auth_type do
            "oauth2" ->
              data
              |> Map.take(["provider", "auth_type", "name", "client_id", "client_secret"])
              |> Map.put("status", "disconnected")

            _ ->
              data
              |> Map.take(["provider", "auth_type", "name"])
              |> Map.put("status", "disconnected")
          end

        save_integration(setting, cleaned)
        Events.broadcast_disconnected(base_provider)
        log_activity("integration.disconnected", base_provider, name, %{}, "manual", actor_uuid)
        :ok

      {:error, _} ->
        :ok
    end
  end

  # ---------------------------------------------------------------------------
  # HTTP helper with auto-refresh
  # ---------------------------------------------------------------------------

  @doc """
  Make an authenticated HTTP request with automatic token refresh on 401.

  For OAuth providers: adds Bearer token, retries with refreshed token on 401.
  For API key providers: adds Bearer token from the api_key.
  For bot token providers: returns credentials for the caller to use directly.

  `opts` are passed through to `Req.request/1`.

  ## Security — `url` must come from a trusted source

  The integration's Bearer token is attached to every request this
  function dispatches. If a caller passes a URL that came from
  unvalidated user input, the token leaks to whatever host that URL
  points at. Callers MUST validate the URL before invoking — pin to
  a domain allowlist (the provider's own host, typically), enforce
  `https`, and reject RFC1918 / loopback / link-local ranges.

  Internal usage (`OpenRouterClient.fetch_models/2`, OAuth refresh,
  userinfo lookups) builds URLs from the Providers registry, which is
  hardcoded and therefore safe. The schema-level `validate_base_url/1`
  guard in `PhoenixKitAI.Endpoint` covers the AI module's
  operator-supplied `base_url` case. New callsites that take URLs
  from anywhere else need their own guard before reaching this
  function — there's no allowlist enforcement here.
  """
  @spec authenticated_request(String.t(), atom(), String.t(), keyword()) ::
          {:ok, Req.Response.t()} | {:error, term()}
  def authenticated_request(uuid, method, url, opts \\ []) when is_binary(uuid) do
    with {:ok, data} <- get_credentials(uuid) do
      token = resolve_bearer_token(data)
      opts = put_auth_header(opts, token)

      case do_request(method, url, opts) do
        {:ok, %{status: 401}} = _unauthorized ->
          retry_with_refreshed_token(uuid, data, method, url, opts)

        other ->
          other
      end
    end
  end

  # ---------------------------------------------------------------------------
  # Listing
  # ---------------------------------------------------------------------------

  @doc """
  List all configured integrations (those that have saved data).

  Order is determined by `load_all_connections/1`'s map iteration,
  which is alphabetical by provider key (Erlang `Map` order over
  string keys) — NOT the registration order from `Providers.all/0`.
  Within a provider, connections are sorted alphabetically by name
  (case-insensitive). Callers that need a specific provider order
  should walk `Providers.all/0` themselves.
  """
  @spec list_integrations() :: [map()]
  def list_integrations do
    provider_keys = Providers.all() |> Enum.map(& &1.key)

    load_all_connections(provider_keys)
    |> Enum.flat_map(fn {_provider, connections} ->
      Enum.map(connections, fn %{data: data} -> data end)
    end)
  end

  @doc """
  List all known providers.
  """
  @spec list_providers() :: [map()]
  def list_providers do
    Providers.all()
  end

  @doc """
  Lists all connections for a provider.

  Returns a list of `%{uuid, name, data, date_added}` maps, sorted by
  name (case-insensitive). Filters by `module = "integrations"` and
  JSONB `provider` — provider and name both live in JSONB; the row's
  `uuid` is the stable identifier. `date_added` is the row's creation
  timestamp (UTC, second precision), useful for "Created N days ago"
  display in UI pickers.
  """
  @spec list_connections(String.t()) :: [
          %{
            uuid: String.t(),
            name: String.t(),
            data: map(),
            date_added: DateTime.t() | nil
          }
        ]
  def list_connections(provider_key) when is_binary(provider_key) do
    PhoenixKit.RepoHelper.repo().all(
      from s in Setting,
        where: s.module == ^@settings_module,
        where: fragment("?->>'provider' = ?", s.value_json, ^provider_key)
    )
    |> Enum.map(&to_connection_map/1)
    |> Enum.sort_by(fn %{name: name} -> String.downcase(name) end)
  end

  defp to_connection_map(%Setting{uuid: uuid, value_json: data, date_added: date_added}) do
    decrypted = Encryption.decrypt_fields(data || %{})

    %{
      uuid: uuid,
      name: Map.get(decrypted, "name", ""),
      data: decrypted,
      date_added: date_added
    }
  end

  @doc """
  Loads all connections for multiple providers in a single database query.

  More efficient than calling `list_connections/1` in a loop. Returns a
  map of `provider_key => [%{uuid, name, data, date_added}]`, with every
  requested provider key present (empty list when no connections exist).
  """
  @spec load_all_connections([String.t()]) :: %{
          String.t() => [
            %{
              uuid: String.t(),
              name: String.t(),
              data: map(),
              date_added: DateTime.t() | nil
            }
          ]
        }
  def load_all_connections(provider_keys) when is_list(provider_keys) do
    all_settings =
      PhoenixKit.RepoHelper.repo().all(
        from s in Setting,
          where: s.module == ^@settings_module,
          where: fragment("?->>'provider' = ANY(?)", s.value_json, ^provider_keys)
      )

    grouped =
      Enum.reduce(all_settings, %{}, fn %Setting{} = setting, acc ->
        conn = to_connection_map(setting)
        provider = conn.data["provider"]

        if is_binary(provider) do
          Map.update(acc, provider, [conn], &[conn | &1])
        else
          acc
        end
      end)

    Map.new(provider_keys, fn pk ->
      connections =
        Map.get(grouped, pk, [])
        |> Enum.sort_by(fn %{name: name} -> String.downcase(name) end)

      {pk, connections}
    end)
  end

  @doc """
  Adds a new named connection for a provider.

  This is the row-birth path. The row's UUIDv7 is generated up front and
  used as both the `uuid` and the `key` column; provider and name live
  in JSONB. Names are pure user-chosen labels with no character or
  uniqueness restrictions — any non-empty string (after trim) is valid,
  duplicates within a provider are allowed.

  Returns `{:ok, %{uuid: uuid, data: data}}` on success.
  """
  @spec add_connection(String.t(), String.t(), String.t() | nil) ::
          {:ok, %{uuid: String.t(), data: map()}}
          | {:error, :empty_name | term()}
  def add_connection(provider_key, name, actor_uuid \\ nil)
      when is_binary(provider_key) and is_binary(name) do
    trimmed = String.trim(name)

    if trimmed == "" do
      {:error, :empty_name}
    else
      data = %{
        "provider" => provider_key,
        "name" => trimmed,
        "auth_type" => provider_auth_type(provider_key),
        "status" => "disconnected"
      }

      case insert_integration_row(data) do
        {:ok, %Setting{uuid: uuid}} ->
          Events.broadcast_connection_added(provider_key, trimmed)

          log_activity(
            "integration.connection_added",
            provider_key,
            trimmed,
            %{},
            "manual",
            actor_uuid
          )

          {:ok, %{uuid: uuid, data: data}}

        {:error, _} = error ->
          error
      end
    end
  end

  # Row-birth path: generate a UUIDv7, use it as both the primary key
  # and the `key` column, and insert with `module = "integrations"`. The
  # `put_change(:uuid, ...)` forces the changeset to use our generated
  # uuid instead of letting Ecto autogenerate a different one.
  defp insert_integration_row(data) do
    uuid = UUIDv7.generate()
    encrypted = Encryption.encrypt_fields(data)

    %Setting{}
    |> Setting.changeset(%{key: uuid, value_json: encrypted, module: @settings_module})
    |> Ecto.Changeset.put_change(:uuid, uuid)
    |> Queries.insert_setting()
  end

  @doc """
  Removes a connection by uuid.

  Names are pure user-chosen labels — no privileged values. The user is
  free to delete any connection; consumer modules that referenced the
  deleted integration row will surface a `:not_configured` (or similar)
  error on next use, which is the correct loud failure.
  """
  @spec remove_connection(String.t(), String.t() | nil) :: :ok | {:error, term()}
  def remove_connection(uuid, actor_uuid \\ nil) when is_binary(uuid) do
    case resolve_uuid(uuid) do
      {:ok, %{provider: provider, name: name, setting: setting}} ->
        case Settings.delete_setting(setting.key) do
          {:ok, _} ->
            Events.broadcast_connection_removed(provider, name)

            log_activity(
              "integration.connection_removed",
              provider,
              name,
              %{},
              "manual",
              actor_uuid
            )

            :ok

          {:error, :not_found} ->
            :ok

          error ->
            error
        end

      {:error, _} ->
        :ok
    end
  end

  @doc """
  Renames a connection identified by uuid.

  Updates only the JSONB `name` field; the storage key (= row uuid) is
  untouched, so consumers that pinned to the uuid keep working across
  the rename. Names are pure user-chosen labels; any non-empty string
  is valid, duplicates within a provider are allowed.

  No-ops when `new_name` (after trim) matches the current name.

  Returns `{:ok, new_data}` on success.
  """
  @spec rename_connection(String.t(), String.t(), String.t() | nil) ::
          {:ok, map()}
          | {:error, :empty_name | :not_configured | term()}
  def rename_connection(uuid, new_name, actor_uuid \\ nil)
      when is_binary(uuid) and is_binary(new_name) do
    trimmed = String.trim(new_name)

    with {:ok, %{provider: provider, name: old_name, setting: setting, data: data}} <-
           resolve_uuid(uuid) do
      cond do
        trimmed == old_name ->
          {:ok, data}

        trimmed == "" ->
          {:error, :empty_name}

        true ->
          updated = Map.put(data, "name", trimmed)

          case save_integration(setting, updated) do
            {:ok, saved} ->
              Events.broadcast_connection_renamed(provider, old_name, trimmed)

              log_activity(
                "integration.connection_renamed",
                provider,
                trimmed,
                %{"old_name" => old_name, "new_name" => trimmed},
                "manual",
                actor_uuid
              )

              {:ok, saved}

            error ->
              error
          end
      end
    end
  end

  # ---------------------------------------------------------------------------
  # Validation
  # ---------------------------------------------------------------------------

  @doc """
  Validate that a provider's credentials are working.

  For OAuth: calls the provider's userinfo endpoint.
  For API key / bot token: calls the provider's validation endpoint if defined.
  Returns `:ok` or `{:error, reason}`.
  """
  @spec validate_connection(String.t(), String.t() | nil) :: :ok | {:error, String.t()}
  def validate_connection(uuid, actor_uuid \\ nil) when is_binary(uuid) do
    {result, log_provider, log_name} =
      case resolve_uuid(uuid) do
        {:ok, %{provider: base_provider, name: name, data: data}} ->
          provider = Providers.get(base_provider)

          inner =
            cond do
              is_nil(provider) -> {:error, gettext("Unknown provider")}
              not has_credentials?(data) -> {:error, gettext("Not configured")}
              true -> do_validate(provider, data)
            end

          {inner, base_provider, name}

        {:error, _} ->
          {{:error, gettext("Not configured")}, "unknown", ""}
      end

    case result do
      :ok ->
        log_activity(
          "integration.validated",
          log_provider,
          log_name,
          %{"result" => "ok"},
          "manual",
          actor_uuid
        )

      {:error, reason} ->
        log_activity(
          "integration.validated",
          log_provider,
          log_name,
          %{"result" => "error", "reason" => reason},
          "manual",
          actor_uuid
        )
    end

    result
  rescue
    # Narrow rescue — only catch exceptions we expect from the
    # validate path so genuine logic bugs (KeyError, ArgumentError,
    # MatchError, etc.) bubble up to the supervisor instead of getting
    # swallowed under a generic "validation failed" message. The three
    # caught classes cover the realistic failure modes:
    #
    #   * `DBConnection.OwnershipError` — sandbox checkout race in
    #     tests that hit this path from an async-shared connection.
    #   * `Postgrex.Error` — DB outage / table-missing on the
    #     `log_activity` write (Activity.log hits `phoenix_kit_activity`
    #     and can raise when the table is unreachable or absent).
    #   * `Req.TransportError` / generic transport — should already
    #     be returned as `{:error, _}` by `check_http/2`; this is the
    #     belt-and-braces case.
    e in [DBConnection.OwnershipError, Postgrex.Error, Req.TransportError] ->
      Logger.error(
        "[Integrations] validate_connection error for #{uuid}: #{Exception.message(e)}"
      )

      {:error, gettext("Validation failed unexpectedly")}
  end

  @doc """
  Probe a provider's API with in-memory credentials, without
  persisting anything. Used by the integration form to let
  operators test what they typed before committing — same HTTP
  validation as `validate_connection/2`, but no storage row, no
  `last_validated_at` stamp, no PubSub broadcast.

  `attrs` is the same shape `save_setup/3` accepts (e.g.
  `%{"api_key" => "..."}` for api_key providers,
  `%{"client_id" => "...", "client_secret" => "..."}` for OAuth).

  OAuth providers without a saved `access_token` will return
  `{:error, "No access token"}` — pre-save validation is most
  useful for api_key / bot_token providers where the secret the
  user just typed IS the credential.
  """
  @spec validate_credentials(String.t(), map()) :: :ok | {:error, String.t()}
  def validate_credentials(provider_key, attrs)
      when is_binary(provider_key) and is_map(attrs) do
    case Providers.get(provider_key) do
      nil -> {:error, gettext("Unknown provider")}
      provider -> do_validate(provider, attrs)
    end
  rescue
    # Mirror the narrow rescue on `validate_connection/2` — same
    # `do_validate/2` codepath, same expected exception surface. Logic
    # bugs (`KeyError`, `ArgumentError`, etc.) must bubble up instead
    # of being masked under a generic "validation failed".
    e in [DBConnection.OwnershipError, Postgrex.Error, Req.TransportError] ->
      Logger.error(
        "[Integrations] validate_credentials error for #{provider_key}: #{Exception.message(e)}"
      )

      {:error, gettext("Validation failed unexpectedly")}
  end

  defp do_validate(%{auth_type: :oauth2} = provider, data) do
    token = data["access_token"]
    config = provider.oauth_config || %{}
    userinfo_url = config[:userinfo_url] || config["userinfo_url"]

    cond do
      not (is_binary(token) and token != "") -> {:error, gettext("No access token")}
      is_nil(userinfo_url) -> :ok
      true -> check_http(userinfo_url, [{"authorization", "Bearer #{token}"}])
    end
  end

  defp do_validate(%{auth_type: auth_type} = provider, data)
       when auth_type in [:api_key, :bot_token] do
    token = data["api_key"] || data["bot_token"] || ""

    cond do
      token == "" ->
        {:error, gettext("No credentials configured")}

      Map.has_key?(provider, :validation) and provider.validation != nil ->
        v = provider.validation
        headers = [{v.auth_header, "#{v.auth_prefix}#{token}"}]
        check_http(v.url, headers)

      true ->
        :ok
    end
  end

  defp do_validate(_, _data), do: :ok

  defp check_http(url, headers) do
    case Req.get(url, headers: headers, receive_timeout: @http_timeout) do
      {:ok, %{status: 200}} -> :ok
      {:ok, %{status: 401}} -> {:error, gettext("Invalid credentials")}
      {:ok, %{status: 403}} -> {:error, gettext("Access denied")}
      {:ok, %{status: status}} -> {:error, gettext("Service error %{status}", status: status)}
      {:error, _reason} -> {:error, gettext("Could not reach the service")}
    end
  end

  @doc """
  Persist the outcome of a connection check (manual or automatic) onto the
  integration record and broadcast a PubSub event when status changes.

  `last_validated_at` is always rewritten — it is the canonical
  "moment of the last validation attempt" timestamp, and a manual
  Test-Connection click that returns the same result must still
  advance the field (otherwise the form's "Last tested N ago" reading
  goes stale). Status and `validation_status` are merged in
  unconditionally too — usually the same value as before, so it's a
  no-op write at the JSONB level. The PubSub broadcast is gated on an
  actual state change so high-frequency automatic paths (e.g. token
  refresh failing on every API call) don't spam listing-LV reloads.
  """
  @spec record_validation(String.t(), :ok | {:error, term()}) :: :ok
  def record_validation(uuid, result) when is_binary(uuid) do
    {new_status, validation_text} = validation_fields(result)

    case resolve_uuid(uuid) do
      {:ok, %{provider: base_provider, setting: setting, data: data}} ->
        status_changed =
          data["status"] != new_status or data["validation_status"] != validation_text

        now_iso = DateTime.utc_now() |> DateTime.to_iso8601()

        base_update = %{
          "status" => new_status,
          "last_validated_at" => now_iso,
          "validation_status" => validation_text
        }

        # `connected_at` tracks the LAST successful connection — it's
        # the timestamp the form's "Connected N ago" line reads from,
        # so the user expects it to bump on every successful re-test
        # (a stuck "Connected 35 minutes ago" after a fresh `:ok`
        # reads as "didn't update even though it connected"). This
        # also matches the OAuth `exchange_code/4` path, which
        # always overwrites `connected_at` on a successful token
        # exchange — keeping the two paths consistent.
        update =
          if result == :ok do
            Map.put(base_update, "connected_at", now_iso)
          else
            base_update
          end

        updated = Map.merge(data, update)

        case save_integration(setting, updated) do
          {:ok, _} ->
            if status_changed, do: Events.broadcast_validated(base_provider, result)
            :ok

          _ ->
            :ok
        end

      {:error, _} ->
        Logger.debug("[Integrations] record_validation skipped — uuid #{inspect(uuid)} not found")

        :ok
    end
  end

  defp validation_fields(:ok), do: {"connected", "ok"}

  defp validation_fields({:error, reason}),
    do: {"error", "error: #{format_validation_reason(reason)}"}

  defp format_validation_reason(reason) when is_binary(reason), do: reason

  defp format_validation_reason({:refresh_failed, status}),
    do: "Token refresh failed (HTTP #{status})"

  defp format_validation_reason(:token_refresh_failed), do: "Token refresh failed"
  defp format_validation_reason(reason), do: inspect(reason)

  defp record_refresh_failure(uuid, reason) do
    reason_text = format_validation_reason(reason)
    record_validation(uuid, {:error, reason_text})

    case resolve_uuid(uuid) do
      {:ok, %{provider: provider, name: name}} ->
        log_activity(
          "integration.token_refresh_failed",
          provider,
          name,
          %{"reason" => reason_text},
          "auto",
          nil
        )

      _ ->
        :ok
    end
  end

  defp maybe_record_recovery(uuid) do
    case resolve_uuid(uuid) do
      {:ok, %{provider: provider, name: name, data: %{"status" => "error"}}} ->
        record_validation(uuid, :ok)
        log_activity("integration.auto_recovered", provider, name, %{}, "auto", nil)

      _ ->
        :ok
    end
  end

  # ---------------------------------------------------------------------------
  # Private
  # ---------------------------------------------------------------------------

  @uuid_pattern ~r/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i

  defp uuid?(str), do: is_binary(str) and Regex.match?(@uuid_pattern, str)

  # Resolve a settings-row uuid to its storage info — single source of
  # truth for the uuid-strict public API. Provider and name are sourced
  # from JSONB; the live `Setting` struct is returned so callers can
  # update in place via changeset.
  defp resolve_uuid(uuid) when is_binary(uuid) and uuid != "" do
    case Queries.get_setting_by_uuid(uuid) do
      %Setting{module: "integrations"} = setting ->
        decrypted = Encryption.decrypt_fields(setting.value_json || %{})

        {:ok,
         %{
           uuid: uuid,
           provider: Map.get(decrypted, "provider", ""),
           name: Map.get(decrypted, "name", ""),
           setting: setting,
           data: decrypted
         }}

      _ ->
        {:error, :not_configured}
    end
  end

  defp resolve_uuid(_), do: {:error, :invalid_uuid}

  defp provider_auth_type(provider_key) do
    case Providers.get(provider_key) do
      %{auth_type: auth_type} -> Atom.to_string(auth_type)
      nil -> nil
    end
  end

  defp save_integration(%Setting{} = setting, data) do
    encrypted_data = Encryption.encrypt_fields(data)

    setting
    |> Setting.update_changeset(%{value_json: encrypted_data, module: @settings_module})
    |> Queries.update_setting()
    |> case do
      {:ok, _setting} ->
        {:ok, data}

      {:error, changeset} = error ->
        Logger.error(
          "[Integrations] Failed to save integration #{setting.uuid}: #{inspect(changeset)}"
        )

        error
    end
  end

  defp fetch_provider(provider_key) do
    case Providers.get(provider_key) do
      nil -> {:error, :unknown_provider}
      provider -> {:ok, provider}
    end
  end

  defp fetch_userinfo_safe(provider, access_token) do
    if provider.oauth_config do
      case OAuth.fetch_userinfo(provider.oauth_config, access_token) do
        {:ok, info} -> info
        _ -> %{}
      end
    else
      %{}
    end
  end

  defp maybe_set_userinfo(data, userinfo) do
    metadata = Map.get(data, "metadata", %{})

    updated_metadata =
      metadata
      |> maybe_put("connected_email", userinfo["email"])
      |> maybe_put("name", userinfo["name"])
      |> maybe_put("picture", userinfo["picture"])

    data
    |> Map.put("metadata", updated_metadata)
    |> maybe_put("external_account_id", userinfo["email"])
    |> maybe_put("external_account_name", userinfo["name"])
  end

  defp maybe_put(map, _key, nil), do: map
  defp maybe_put(map, _key, ""), do: map
  defp maybe_put(map, key, value), do: Map.put(map, key, value)

  defp maybe_set_status(data, nil), do: Map.put_new(data, "status", "disconnected")

  defp maybe_set_status(data, provider) do
    # If already validated as "connected" or "error", keep the status
    # Only set status on initial save
    if data["status"] in ["connected", "error"] do
      data
    else
      has_creds =
        case provider.auth_type do
          :oauth2 -> has_token?(data)
          :api_key -> present?(data["api_key"])
          :bot_token -> present?(data["bot_token"])
          :key_secret -> present?(data["access_key"])
          :credentials -> has_custom_creds?(data) or has_flat_credential_fields?(provider, data)
        end

      # Saved-but-not-validated. Setting "connected" here was optimistic —
      # nothing has actually been tested. Callers that want a real
      # `connected` status should follow up with `validate_connection/2`
      # and `record_validation/2`. The form LV does this automatically on
      # the save_setup / create_connection flow.
      if has_creds do
        Map.put(data, "status", "configured")
      else
        Map.put(data, "status", "disconnected")
      end
    end
  end

  defp maybe_set_connected_at(data) do
    if data["status"] == "connected" and is_nil(data["connected_at"]) do
      Map.put(data, "connected_at", DateTime.utc_now() |> DateTime.to_iso8601())
    else
      data
    end
  end

  defp has_credentials?(%{"status" => status}) when status in ["connected", "configured"],
    do: true

  defp has_credentials?(data),
    do:
      present?(data["access_token"]) or present?(data["api_key"]) or present?(data["bot_token"]) or
        present?(data["access_key"]) or has_custom_creds?(data) or
        has_flat_credential_fields?(Providers.get(data["provider"]), data)

  defp has_custom_creds?(%{"credentials" => creds}) when is_map(creds) and map_size(creds) > 0,
    do: true

  defp has_custom_creds?(_), do: false

  # `:credentials` providers (e.g. universal SMTP) store their fields flat
  # on the data map — there's no privileged nested "credentials" shape to
  # check for them (that's `has_custom_creds?/1`, used by consumers that
  # actually nest their creds). Instead, treat the connection as having
  # credentials once every *required* setup field the provider declares is
  # present. Data-driven off the provider's own field list, so it applies
  # to any `:credentials` provider without hardcoding field names here.
  defp has_flat_credential_fields?(%{auth_type: :credentials, setup_fields: fields}, data) do
    required = Enum.filter(fields, & &1.required)

    # Guard the empty-required-list footgun: `Enum.all?([], _)` is `true`, which
    # would treat a credentials provider with no required fields as "configured".
    required != [] and
      Enum.all?(required, fn %{key: key} -> field_present?(data[key]) end)
  end

  defp has_flat_credential_fields?(_provider, _data), do: false

  # `:number` setup fields (e.g. SMTP `port`) may arrive as an integer rather
  # than a string, which `present?/1` (binary-only) would wrongly reject.
  defp field_present?(val) when is_number(val), do: true
  defp field_present?(val), do: present?(val)

  defp present?(val), do: is_binary(val) and val != ""

  defp has_token?(data), do: present?(data["access_token"])

  defp resolve_bearer_token(data) do
    data["access_token"] || data["api_key"] || data["bot_token"] || ""
  end

  defp put_auth_header(opts, token) do
    headers =
      opts
      |> Keyword.get(:headers, [])
      |> Enum.reject(fn {k, _} -> String.downcase(to_string(k)) == "authorization" end)

    headers = [{"authorization", "Bearer #{token}"} | headers]
    Keyword.put(opts, :headers, headers)
  end

  defp do_request(method, url, opts) do
    Req.request([method: method, url: url] ++ opts)
  end

  defp retry_with_refreshed_token(uuid, data, method, url, opts) do
    if data["auth_type"] == "oauth2" and is_binary(data["refresh_token"]) and
         data["refresh_token"] != "" do
      case refresh_access_token(uuid) do
        {:ok, new_token} ->
          opts = put_auth_header(opts, new_token)
          do_request(method, url, opts)

        {:error, reason} ->
          Logger.warning("[Integrations] Token refresh failed for #{uuid}: #{inspect(reason)}")

          {:error, :token_refresh_failed}
      end
    else
      Logger.warning("[Integrations] 401 for #{uuid} but no refresh_token available")
      {:error, :unauthorized}
    end
  end

  # ---------------------------------------------------------------------------
  # Activity logging
  # ---------------------------------------------------------------------------

  defp log_activity(action, provider, name, metadata, mode, actor_uuid)
       when is_binary(provider) and is_binary(name) do
    if Code.ensure_loaded?(PhoenixKit.Activity) do
      # Stamp the storage row's uuid so the activity deep-links to the
      # connection's edit page (via PhoenixKit.Integrations.ResourceLinks).
      # nil when the row can't be resolved (e.g. a disconnect that removed it) —
      # the feed still renders, just without a link.
      resource_uuid =
        case find_uuid_by_provider_name({provider, name}) do
          {:ok, uuid} -> uuid
          _ -> nil
        end

      PhoenixKit.Activity.log(%{
        action: action,
        module: "integrations",
        mode: mode,
        actor_uuid: actor_uuid,
        resource_type: "integration",
        resource_uuid: resource_uuid,
        metadata:
          Map.merge(metadata, %{
            "provider" => provider,
            "connection" => name,
            "actor_role" => "admin"
          })
      })
    end
  rescue
    e ->
      Logger.warning("[Integrations] Failed to log activity #{action}: #{Exception.message(e)}")
  end

  # ---------------------------------------------------------------------------
  # Legacy migration (deprecated entry point)
  # ---------------------------------------------------------------------------

  @doc """
  Deprecated. Use `PhoenixKit.ModuleRegistry.run_all_legacy_migrations/0`
  from your host app's `Application.start/2` instead.

  Each module that has legacy data now implements its own
  `migrate_legacy/0` callback. The orchestrator walks every registered
  module and runs them all — same single entry point as before, but
  modules own their own data shape.

  Calling this delegates to the orchestrator for backwards compat.
  Returns `:ok` regardless of per-module outcome (matches the previous
  semantics of "best-effort, never crash boot").
  """
  @deprecated "Use PhoenixKit.ModuleRegistry.run_all_legacy_migrations/0 instead"
  @spec run_legacy_migrations() :: :ok
  def run_legacy_migrations do
    _ = PhoenixKit.ModuleRegistry.run_all_legacy_migrations()
    :ok
  end
end