towerops/lib/towerops/devices.ex

defmodule Towerops.Devices do
  @moduledoc """
  The Devices context.
  """

  import Ecto.Query

  alias Towerops.Agents
  alias Towerops.Devices.CredentialResolver
  alias Towerops.Devices.Device, as: DeviceSchema
  alias Towerops.Devices.Event
  alias Towerops.Monitoring
  alias Towerops.Organizations
  alias Towerops.Organizations.SubscriptionLimits
  alias Towerops.Repo
  alias Towerops.Sites
  alias Towerops.Workers.DeviceMonitorWorker
  alias Towerops.Workers.DiscoveryWorker

  @doc """
  Returns the list of devices for a site.
  Ordered by custom display_order (if set), then alphabetically by name.
  """
  @spec list_site_devices(String.t()) :: [DeviceSchema.t()]
  def list_site_devices(site_id) do
    Repo.all(
      from(e in DeviceSchema,
        where: e.site_id == ^site_id,
        order_by: [asc: e.display_order, asc: e.name]
      )
    )
  end

  @doc """
  Returns the list of all devices for an organization.
  Ordered by custom display_order (if set), then alphabetically by name.

  Supports filtering by:
  - site_id: Filter by specific site
  - status: Filter by status ("up", "down", "unknown")
  """
  @spec list_organization_devices(String.t(), map()) :: [DeviceSchema.t()]
  def list_organization_devices(organization_id, filters \\ %{}) do
    query =
      from(e in DeviceSchema,
        left_join: s in assoc(e, :site),
        where: e.organization_id == ^organization_id,
        order_by: [asc: e.display_order, asc: e.name],
        preload: [site: s]
      )

    query =
      if site_id = filters["site_id"] do
        where(query, [e], e.site_id == ^site_id)
      else
        query
      end

    query =
      if status = filters["status"] do
        where(query, [e], e.status == ^status)
      else
        query
      end

    Repo.all(query)
  end

  @doc """
  Returns all devices for multiple organizations (for GDPR data access).
  """
  @spec list_devices_for_organizations([String.t()]) :: [DeviceSchema.t()]
  def list_devices_for_organizations(organization_ids) when is_list(organization_ids) do
    Repo.all(
      from(e in DeviceSchema,
        left_join: s in assoc(e, :site),
        join: o in assoc(e, :organization),
        where: e.organization_id in ^organization_ids,
        order_by: [asc: e.organization_id, asc: e.name],
        preload: [site: s, organization: o]
      )
    )
  end

  @doc """
  Returns the count of devices for an organization.
  """
  @spec count_organization_devices(String.t()) :: integer()
  def count_organization_devices(organization_id) do
    Repo.aggregate(
      from(e in DeviceSchema,
        where: e.organization_id == ^organization_id
      ),
      :count
    )
  end

  @doc """
  Returns the count of devices for a site.
  """
  @spec count_site_devices(String.t()) :: integer()
  def count_site_devices(site_id) do
    Repo.aggregate(
      from(e in DeviceSchema, where: e.site_id == ^site_id),
      :count
    )
  end

  @doc """
  Returns the count of devices that is down for a site.
  """
  def count_site_devices_down(site_id) do
    Repo.aggregate(
      from(e in DeviceSchema, where: e.site_id == ^site_id and e.status == :down),
      :count
    )
  end

  @doc """
  Search devices by name or IP address for an organization.
  Returns up to 10 results. Requires at least 2 characters.
  """
  @spec search_devices(String.t(), String.t()) :: [DeviceSchema.t()]
  def search_devices(organization_id, query) do
    search_term = "%#{query}%"

    DeviceSchema
    |> where(organization_id: ^organization_id)
    |> where([d], ilike(d.name, ^search_term) or ilike(d.ip_address, ^search_term))
    |> order_by(:name)
    |> limit(10)
    |> Repo.all()
    |> Repo.preload(:site)
  end

  @doc """
  Returns a map of device status counts for an organization.

  Uses a GROUP BY query instead of loading all devices into memory.

  ## Examples

      iex> get_device_status_counts(organization_id)
      %{up: 5, down: 2, unknown: 1}

      iex> get_device_status_counts(empty_org_id)
      %{}
  """
  @spec get_device_status_counts(String.t()) :: %{atom() => non_neg_integer()}
  def get_device_status_counts(organization_id) do
    from(d in DeviceSchema,
      where: d.organization_id == ^organization_id,
      group_by: d.status,
      select: {d.status, count(d.id)}
    )
    |> Repo.all()
    |> Map.new()
  end

  @doc """
  Returns the list of all devices with monitoring enabled.
  """
  def list_monitored_devices do
    Repo.all(
      from(e in DeviceSchema,
        where: e.monitoring_enabled == true,
        order_by: [asc: e.name]
      )
    )
  end

  @doc """
  Returns the list of all devices with SNMP enabled.
  """
  def list_snmp_enabled_devices do
    Repo.all(
      from(e in DeviceSchema,
        where: e.snmp_enabled == true,
        order_by: [asc: e.name]
      )
    )
  end

  @doc """
  Lists all devices with MikroTik API enabled and configured for backup.
  Returns devices that have MikroTik API enabled, credentials, and an assigned agent.
  """
  def list_mikrotik_devices_with_api do
    # Get all MikroTik devices with API enabled and credentials configured
    # Credentials are denormalized - already resolved on device during create/update
    # Agent assignment is resolved hierarchically in backup_device/1 (includes global default)
    Repo.all(
      from(d in DeviceSchema,
        left_join: s in assoc(d, :site),
        left_join: o in assoc(d, :organization),
        left_join: aa in assoc(d, :agent_assignments),
        where: d.mikrotik_enabled == true,
        where: not is_nil(d.mikrotik_username),
        order_by: [asc: d.name],
        preload: [site: s, organization: o, agent_assignments: aa],
        distinct: d.id
      )
    )
  end

  @doc """
  Resolves the agent token ID for a device using hierarchical resolution:
  1. Direct agent assignment (via agent_assignments table)
  2. Site-level agent assignment
  3. Organization default agent
  4. Global default cloud poller

  Returns nil if no agent is assigned.
  """
  def resolve_agent_token_id(device) do
    # Ensure organization, site (if present), and agent_assignments are preloaded
    # Note: site: :organization is needed for fallback agent token resolution
    device =
      if device.site_id do
        Repo.preload(device, [:organization, [site: :organization], agent_assignments: []])
      else
        Repo.preload(device, [:organization, agent_assignments: []])
      end

    # Try each level in order
    get_direct_agent(device) ||
      get_site_agent(device) ||
      get_org_default_agent(device) ||
      get_global_default_agent()
  end

  defp get_direct_agent(device) do
    case Enum.find(device.agent_assignments, & &1.enabled) do
      %{agent_token_id: id} when not is_nil(id) -> id
      _ -> nil
    end
  end

  defp get_site_agent(device) do
    # Safe: only access site if it exists
    if device.site_id && device.site do
      device.site.agent_token_id
    end
  end

  defp get_org_default_agent(device) do
    # Use direct organization relationship instead of site.organization
    device.organization.default_agent_token_id
  end

  defp get_global_default_agent, do: Towerops.Settings.get_global_default_cloud_poller()

  @doc """
  Gets a single device by ID, returns nil if not found.
  """
  def get_device(id) do
    DeviceSchema
    |> Repo.get(id)
    |> case do
      nil -> nil
      device -> Repo.preload(device, [:organization, site: :organization])
    end
  end

  @doc """
  Gets a single device by ID, raises if not found.
  """
  def get_device!(id) do
    DeviceSchema
    |> Repo.get!(id)
    |> Repo.preload([:organization, site: :organization])
  end

  @doc """
  Gets device with full details including SNMP device, interfaces, and sensors.
  """
  def get_device_with_details(id) do
    DeviceSchema
    |> Repo.get(id)
    |> case do
      nil ->
        nil

      device ->
        Repo.preload(device, [:site, :organization, snmp_device: [:interfaces, :sensors]])
    end
  end

  @doc """
  Finds a device by IP address within a site, optionally excluding a specific device ID.
  Returns nil if no device found.
  """
  def get_device_by_ip(ip_address, site_id, exclude_id \\ nil)

  def get_device_by_ip(_ip_address, nil, _exclude_id), do: nil

  def get_device_by_ip(ip_address, site_id, exclude_id) do
    query =
      from(d in DeviceSchema,
        where: d.ip_address == ^ip_address and d.site_id == ^site_id,
        preload: [:site]
      )

    query =
      if exclude_id do
        from(d in query, where: d.id != ^exclude_id)
      else
        query
      end

    Repo.one(query)
  end

  @doc """
  Gets a single device belonging to a specific site.
  """
  def get_site_device!(site_id, device_id) do
    Repo.one!(from(e in DeviceSchema, where: e.id == ^device_id, where: e.site_id == ^site_id, preload: [:site]))
  end

  @doc """
  Gets SNMP configuration for device with hierarchical fallback.

  Falls back in this order:
  1. Device-level configuration
  2. Site-level configuration
  3. Organization-level configuration

  Returns a map with:
  - version: SNMP version ("1", "2c", or "3")
  - community: SNMP community string
  - source: Where the config came from (:device, :site, or :organization)
  """
  def get_snmp_config(device_id) when is_binary(device_id) do
    device = Repo.get!(DeviceSchema, device_id)
    get_snmp_config(device)
  end

  def get_snmp_config(%DeviceSchema{} = device) do
    # Credentials are already denormalized on the device - just read them!
    %{
      version: device.snmp_version || "2c",
      community: device.snmp_community,
      port: device.snmp_port || 161,
      transport: device.snmp_transport || "udp",
      source: safe_source_to_atom(device.snmp_community_source)
    }
  end

  # Safely convert credential source to atom with whitelist to prevent atom exhaustion
  defp safe_source_to_atom("device"), do: :device
  defp safe_source_to_atom("site"), do: :site
  defp safe_source_to_atom("organization"), do: :organization
  defp safe_source_to_atom(nil), do: :organization
  defp safe_source_to_atom(_unknown), do: :organization

  @doc """
  Gets MikroTik API configuration for a device with hierarchical fallback.

  Returns a map with username, password (decrypted), port, use_ssl, enabled, and source.

  ## Examples

      iex> get_mikrotik_config(device)
      %{
        username: "admin",
        password: "secret",
        port: 8729,
        use_ssl: true,
        enabled: true,
        source: :device
      }
  """
  def get_mikrotik_config(device_id) when is_binary(device_id) do
    device = Repo.get!(DeviceSchema, device_id)
    get_mikrotik_config(device)
  end

  def get_mikrotik_config(%DeviceSchema{} = device) do
    resolve_mikrotik_config(device)
  end

  defp resolve_mikrotik_config(device) do
    # Credentials are already denormalized on the device - just read them!
    %{
      username: device.mikrotik_username,
      password: device.mikrotik_password,
      port: device.mikrotik_port || 8729,
      ssh_port: device.mikrotik_ssh_port || 22,
      use_ssl: if(device.mikrotik_use_ssl == nil, do: true, else: device.mikrotik_use_ssl),
      enabled: device.mikrotik_enabled || false,
      source: credential_source_atom(device.mikrotik_credential_source)
    }
  end

  @doc """
  Propagates MikroTik credential changes from a site to all devices in that
  site that inherit credentials from the site (credential_source = "site").

  Updates username, password, port, use_ssl, and enabled fields.
  """
  def propagate_site_mikrotik_change(site_id, attrs) do
    # Find all devices in this site that inherit credentials from site
    device_query =
      from(d in DeviceSchema,
        where: d.site_id == ^site_id,
        where: d.mikrotik_credential_source == "site"
      )

    device_ids = Repo.all(from(d in device_query, select: d.id))
    now = DateTime.truncate(DateTime.utc_now(), :second)
    updates = attrs |> Map.new() |> Map.put(:updated_at, now) |> Map.to_list()

    Repo.update_all(device_query, set: updates)

    Enum.each(device_ids, fn device_id ->
      # Broadcast assignment change to trigger agent job list refresh
      Phoenix.PubSub.broadcast(
        Towerops.PubSub,
        "device:#{device_id}:assignments",
        {:assignments_changed, :mikrotik_updated}
      )
    end)

    :ok
  end

  @doc """
  Propagates MikroTik credential changes from an organization to all devices
  in sites that have no site-level credentials (credential_source = "site"
  and site.mikrotik_username is nil).

  Updates username, password, port, use_ssl, and enabled fields.
  """
  def propagate_organization_mikrotik_change(organization_id, attrs) do
    # With denormalized credentials, update all devices that inherit from the organization
    # (devices with mikrotik_credential_source = "organization")
    device_query =
      from(d in DeviceSchema,
        where: d.organization_id == ^organization_id,
        where: d.mikrotik_credential_source == "organization"
      )

    device_ids = Repo.all(from(d in device_query, select: d.id))
    now = DateTime.truncate(DateTime.utc_now(), :second)
    updates = attrs |> Map.new() |> Map.put(:updated_at, now) |> Map.to_list()

    Repo.update_all(device_query, set: updates)

    Enum.each(device_ids, fn device_id ->
      # Broadcast assignment change to trigger agent job list refresh
      Phoenix.PubSub.broadcast(
        Towerops.PubSub,
        "device:#{device_id}:assignments",
        {:assignments_changed, :mikrotik_updated}
      )
    end)

    :ok
  end

  @doc """
  Gets SNMPv3 configuration for a device with hierarchical fallback.

  Returns a map with all v3 fields resolved, or nil if v2c/v1 is in use.

  ## Examples

      iex> get_snmpv3_config(device)
      %{
        security_level: "authPriv",
        username: "snmpuser",
        auth_protocol: "SHA-256",
        auth_password: "decrypted_auth_pass",
        priv_protocol: "AES",
        priv_password: "decrypted_priv_pass",
        source: :device
      }
  """
  def get_snmpv3_config(device_id) when is_binary(device_id) do
    device = Repo.get!(DeviceSchema, device_id)
    get_snmpv3_config(device)
  end

  def get_snmpv3_config(%DeviceSchema{snmp_version: version} = _device) when version != "3" do
    nil
  end

  def get_snmpv3_config(%DeviceSchema{} = device) do
    # Credentials are already denormalized on the device - just read them!
    %{
      security_level: device.snmpv3_security_level,
      username: device.snmpv3_username,
      auth_protocol: device.snmpv3_auth_protocol || "SHA-256",
      auth_password: device.snmpv3_auth_password,
      priv_protocol: device.snmpv3_priv_protocol || "AES",
      priv_password: device.snmpv3_priv_password,
      source: credential_source_atom(device.snmpv3_credential_source)
    }
  end

  @doc """
  Propagates SNMPv3 credential changes from a site to all devices in that
  site that inherit credentials from the site (credential_source = "site").
  """
  def propagate_site_snmpv3_change(site_id, attrs) do
    device_query =
      from(d in DeviceSchema,
        where: d.site_id == ^site_id,
        where: d.snmpv3_credential_source == "site"
      )

    device_ids = Repo.all(from(d in device_query, select: d.id))
    now = DateTime.truncate(DateTime.utc_now(), :second)
    updates = attrs |> Map.new() |> Map.put(:updated_at, now) |> Map.to_list()

    Repo.update_all(device_query, set: updates)

    Enum.each(device_ids, fn device_id ->
      Phoenix.PubSub.broadcast(
        Towerops.PubSub,
        "device:#{device_id}:assignments",
        {:assignments_changed, :snmpv3_updated}
      )
    end)

    :ok
  end

  @doc """
  Propagates SNMPv3 credential changes from an organization to all devices
  in sites that have no site-level credentials.
  """
  def propagate_organization_snmpv3_change(organization_id, attrs) do
    # With denormalized credentials, update all devices that inherit from the organization
    # (devices with snmpv3_credential_source = "organization")
    device_query =
      from(d in DeviceSchema,
        where: d.organization_id == ^organization_id,
        where: d.snmpv3_credential_source == "organization"
      )

    device_ids = Repo.all(from(d in device_query, select: d.id))
    now = DateTime.truncate(DateTime.utc_now(), :second)
    updates = attrs |> Map.new() |> Map.put(:updated_at, now) |> Map.to_list()

    Repo.update_all(device_query, set: updates)

    Enum.each(device_ids, fn device_id ->
      Phoenix.PubSub.broadcast(
        Towerops.PubSub,
        "device:#{device_id}:assignments",
        {:assignments_changed, :snmpv3_updated}
      )
    end)

    :ok
  end

  @doc """
  Creates device.

  ## Options
    * `:bypass_limits` - When true, bypasses subscription device limits (for superusers)
  """
  def create_device(attrs, opts \\ []) do
    bypass_limits = Keyword.get(opts, :bypass_limits, false)

    # Check device quota before insertion (unless bypassing)
    with {:ok, changeset} <- check_device_quota(attrs, bypass_limits),
         {:ok, device} <- Repo.insert(changeset) do
      # Start monitoring/polling if enabled
      _ =
        if device.monitoring_enabled do
          DeviceMonitorWorker.start_monitoring(device.id)
        end

      # Note: Checks will be created and scheduled when discovery runs
      # No need to start polling here

      broadcast_device_change(device.organization_id, :device_created)

      {:ok, device}
    end
  end

  defp check_device_quota(attrs, bypass_limits) do
    changeset = DeviceSchema.changeset(%DeviceSchema{}, attrs)

    # Apply credential resolution
    changeset = apply_credential_resolution(changeset)

    if bypass_limits do
      {:ok, changeset}
    else
      do_check_quota(changeset)
    end
  end

  defp apply_credential_resolution(changeset) do
    # Extract organization_id and site_id from changeset
    organization_id = Ecto.Changeset.get_field(changeset, :organization_id)
    site_id = Ecto.Changeset.get_field(changeset, :site_id)

    if organization_id do
      # Load organization and site (if present)
      organization = Organizations.get_organization!(organization_id)
      site = if site_id, do: Sites.get_site!(site_id)

      # Apply credential resolution
      changeset
      |> CredentialResolver.resolve_snmp_credentials(organization, site)
      |> CredentialResolver.resolve_snmpv3_credentials(organization, site)
      |> CredentialResolver.resolve_mikrotik_credentials(organization, site)
    else
      # No organization_id yet, skip credential resolution
      changeset
    end
  end

  defp do_check_quota(changeset) do
    case Ecto.Changeset.fetch_change(changeset, :organization_id) do
      {:ok, organization_id} ->
        validate_device_quota(changeset, organization_id)

      :error ->
        # No organization_id in changeset, let normal validation handle it
        {:ok, changeset}
    end
  end

  defp validate_device_quota(changeset, organization_id) do
    organization = Organizations.get_organization!(organization_id)

    case SubscriptionLimits.check_device_limit(organization) do
      {:ok, :within_limit} ->
        {:ok, changeset}

      {:error, :at_limit, _current, max} ->
        error_changeset =
          Ecto.Changeset.add_error(
            changeset,
            :base,
            "You've reached your plan limit of #{max} devices. Upgrade to add more."
          )

        {:error, error_changeset}
    end
  end

  @doc """
  Updates device.
  """
  def update_device(%DeviceSchema{} = device, attrs) do
    old_monitoring = device.monitoring_enabled
    old_snmp = device.snmp_enabled
    old_snmp_version = device.snmp_version
    old_snmp_port = device.snmp_port

    changeset = DeviceSchema.changeset(device, attrs)

    # Apply credential resolution
    changeset = apply_credential_resolution(changeset)

    # Validate MikroTik SSL requirement for cloud pollers
    changeset = validate_mikrotik_ssl_with_cloud_poller(changeset, device)

    case Repo.update(changeset) do
      {:ok, updated_device} = result ->
        _ = handle_monitoring_changes(updated_device, old_monitoring)
        _ = handle_snmp_changes(updated_device, old_snmp, old_snmp_version, old_snmp_port)
        broadcast_device_change(updated_device.organization_id, :device_updated)
        result

      error ->
        error
    end
  end

  defp validate_mikrotik_ssl_with_cloud_poller(changeset, device) do
    # Check if MikroTik is enabled and SSL is being disabled
    mikrotik_enabled = Ecto.Changeset.get_field(changeset, :mikrotik_enabled)
    use_ssl_change = Ecto.Changeset.get_change(changeset, :mikrotik_use_ssl)

    # Only validate if MikroTik is enabled and SSL is explicitly being set to false
    if mikrotik_enabled && use_ssl_change == false do
      validate_cloud_poller_requires_ssl(changeset, device)
    else
      changeset
    end
  end

  defp validate_cloud_poller_requires_ssl(changeset, device) do
    # Get effective agent to check if it's a cloud poller
    device_with_site = Repo.preload(device, [:organization, site: :organization])
    {effective_agent_id, _source} = Agents.get_effective_agent_token_with_source(device_with_site)

    using_cloud_poller =
      if effective_agent_id do
        agent = Agents.get_agent_token!(effective_agent_id)
        agent.is_cloud_poller
      else
        # No agent means cloud polling by default
        true
      end

    if using_cloud_poller do
      Ecto.Changeset.add_error(
        changeset,
        :mikrotik_use_ssl,
        "SSL is required when using cloud pollers. Plain MikroTik API sends credentials unencrypted over the internet."
      )
    else
      changeset
    end
  end

  @doc """
  Resolves the effective SNMP community string for a device.

  Resolution order:
  1. If device has its own community string (source = "device" AND community is non-empty), use it
  2. Otherwise, inherit from site's community string
  3. If site has no community, inherit from organization's community string
  4. If no community found anywhere, return nil

  Note: Even if source is "device", we fall back to inheritance if the device community is empty.
  This handles cases where the source field is incorrect or the community was cleared.
  """
  def resolve_snmp_community(%DeviceSchema{} = device) do
    device = Repo.preload(device, [:site, :organization, site: :organization])

    community =
      get_device_community(device) ||
        get_site_community(device) ||
        get_organization_community(device)

    log_empty_community(device, community)
    community
  end

  # Get device-specific community if source is "device" and value is present
  defp get_device_community(device) do
    if device.snmp_community_source == "device" && present?(device.snmp_community) do
      device.snmp_community
    end
  end

  # Get community from site (handles nil site)
  defp get_site_community(%{site: nil}), do: nil
  defp get_site_community(%{site: site}), do: site.snmp_community

  # Get community from organization
  defp get_organization_community(%{organization: organization}) do
    organization.snmp_community
  end

  # Log warning if community is empty
  defp log_empty_community(device, community) when community in [nil, ""] do
    require Logger

    Logger.warning("Empty SNMP community resolved for device",
      device_id: device.id,
      device_name: device.name,
      community_source: device.snmp_community_source,
      device_community: device.snmp_community,
      site_community: get_site_community(device),
      org_community: device.organization.snmp_community,
      site_id: device.site_id,
      org_id: device.organization_id
    )
  end

  defp log_empty_community(_device, _community), do: :ok

  # Helper to check if a string is present (not nil and not empty)
  defp present?(value) when is_binary(value), do: String.trim(value) != ""
  defp present?(_), do: false

  defp credential_source_atom(value) do
    case value do
      "device" -> :device
      "site" -> :site
      "organization" -> :organization
      _ -> :organization
    end
  end

  @doc """
  Propagates SNMP community string changes from a site to all devices
  that inherit from it (source = "site").

  This updates the actual snmp_community value on devices so they have
  the resolved value cached and ready for agent communication.
  """
  def propagate_site_community_change(site_id, new_community) do
    # Find all devices in this site that inherit community from site
    device_query =
      from(d in DeviceSchema, where: d.site_id == ^site_id, where: d.snmp_community_source == "site")

    device_ids = Repo.all(from(d in device_query, select: d.id))
    now = DateTime.truncate(DateTime.utc_now(), :second)

    Repo.update_all(device_query, set: [snmp_community: new_community, updated_at: now])

    Enum.each(device_ids, fn device_id ->
      # Broadcast assignment change to trigger agent job list refresh
      Phoenix.PubSub.broadcast(
        Towerops.PubSub,
        "device:#{device_id}:assignments",
        {:assignments_changed, :community_updated}
      )
    end)

    :ok
  end

  @doc """
  Propagates SNMP community string changes from an organization to all
  devices in sites that have no site-level community (source = "site"
  and site.snmp_community is nil).
  """
  def propagate_organization_community_change(organization_id, new_community) do
    # With denormalized credentials, update all devices that inherit from the organization
    # (devices with snmp_community_source = "organization")
    device_query =
      from(d in DeviceSchema,
        where: d.organization_id == ^organization_id,
        where: d.snmp_community_source == "organization"
      )

    device_ids = Repo.all(from(d in device_query, select: d.id))
    now = DateTime.truncate(DateTime.utc_now(), :second)

    Repo.update_all(device_query, set: [snmp_community: new_community, updated_at: now])

    Enum.each(device_ids, fn device_id ->
      # Broadcast assignment change to trigger agent job list refresh
      Phoenix.PubSub.broadcast(
        Towerops.PubSub,
        "device:#{device_id}:assignments",
        {:assignments_changed, :community_updated}
      )
    end)

    :ok
  end

  @doc """
  Deletes device.
  """
  def delete_device(%DeviceSchema{} = device) do
    organization_id = device.organization_id

    # Stop monitoring and check jobs before deleting
    _ = DeviceMonitorWorker.stop_monitoring(device.id)
    _ = Monitoring.stop_device_checks(device.id)

    # Wait briefly for any in-flight jobs to complete or detect deletion
    # This reduces (but doesn't eliminate) the window for orphaned writes
    # In-flight jobs check device existence via verify_polling_assignment_unchanged
    Process.sleep(500)

    # Get agent assignment before deleting (if any)
    assignment = Agents.get_device_assignment(device.id)
    agent_token_id = if assignment, do: assignment.agent_token_id

    # Delete the device (cascades to SNMP device, sensors, interfaces, etc.)
    result = Repo.delete(device)

    case result do
      {:ok, _} ->
        # Notify agent to stop processing jobs for this device
        if agent_token_id do
          Agents.broadcast_assignment_change(agent_token_id, :device_deleted)
        end

        broadcast_device_change(organization_id, :device_deleted)

      _ ->
        # Notify agent even on failure (preserves original behavior)
        if agent_token_id do
          Agents.broadcast_assignment_change(agent_token_id, :device_deleted)
        end
    end

    result
  end

  @doc """
  Returns an `%Ecto.Changeset{}` for tracking device changes.
  """
  def change_device(%DeviceSchema{} = device, attrs \\ %{}) do
    DeviceSchema.changeset(device, attrs)
  end

  @doc """
  Updates the status of device.
  """
  def update_device_status(%DeviceSchema{} = device, new_status) do
    now = DateTime.truncate(DateTime.utc_now(), :second)
    status_changed = device.status != new_status

    changes = %{
      status: new_status,
      last_checked_at: now
    }

    changes =
      if status_changed do
        Map.put(changes, :last_status_change_at, now)
      else
        changes
      end

    case device |> Ecto.Changeset.change(changes) |> Repo.update() do
      {:ok, updated_device} = result ->
        if status_changed do
          broadcast_device_change(updated_device.organization_id, :device_status_changed)
        else
          broadcast_device_change(updated_device.organization_id, :device_updated)
        end

        result

      error ->
        error
    end
  end

  @doc """
  Updates the last SNMP poll timestamp for device.
  Used for distributed coordination to prevent duplicate polling across pods.
  """
  def update_snmp_poll_time(%DeviceSchema{} = device) do
    now = DateTime.truncate(DateTime.utc_now(), :second)

    device
    |> Ecto.Changeset.change(%{last_snmp_poll_at: now})
    |> Repo.update()
  end

  # Private helpers for monitoring/polling management

  defp handle_monitoring_changes(device, old_monitoring) do
    cond do
      device.monitoring_enabled && !old_monitoring ->
        _ = DeviceMonitorWorker.start_monitoring(device.id)
        # Trigger discovery when monitoring is enabled (if SNMP is also enabled)
        if device.snmp_enabled, do: DiscoveryWorker.enqueue(device.id)

      !device.monitoring_enabled && old_monitoring ->
        _ = DeviceMonitorWorker.stop_monitoring(device.id)

      true ->
        :ok
    end
  end

  defp handle_snmp_changes(device, old_snmp, old_snmp_version, old_snmp_port) do
    should_discover = should_trigger_discovery?(device, old_snmp, old_snmp_version, old_snmp_port)

    cond do
      device.snmp_enabled && !old_snmp ->
        # SNMP newly enabled - discovery will create and schedule checks
        if should_discover, do: DiscoveryWorker.enqueue(device.id)

      !device.snmp_enabled && old_snmp ->
        # SNMP disabled - disable all checks and cancel jobs
        _ = Monitoring.disable_device_checks(device.id)

      should_discover ->
        # SNMP settings changed - re-run discovery
        _ = DiscoveryWorker.enqueue(device.id)

      true ->
        :ok
    end
  end

  defp should_trigger_discovery?(device, old_snmp, old_snmp_version, old_snmp_port) do
    cond do
      # SNMP was just enabled
      device.snmp_enabled && !old_snmp ->
        true

      # SNMP version changed while enabled
      device.snmp_enabled && device.snmp_version != old_snmp_version ->
        true

      # SNMP port changed while enabled
      device.snmp_enabled && device.snmp_port != old_snmp_port ->
        true

      # Otherwise no discovery needed
      true ->
        false
    end
  end

  ## Events

  @doc """
  Creates an device event.
  """
  def create_event(attrs) do
    %Event{}
    |> Event.changeset(attrs)
    |> Repo.insert()
  end

  @doc """
  Returns the list of events for device, ordered by most recent first.
  """
  def list_devices_events(device_id, limit \\ 100) do
    Repo.all(
      from(e in Event,
        where: e.device_id == ^device_id,
        order_by: [desc: e.occurred_at],
        limit: ^limit
      )
    )
  end

  ## Display Order Management

  @doc """
  Reorders a device to a new position within its site.

  Updates display_order for all devices in the site to maintain continuous ordering (1, 2, 3, ...).
  The new_position is 1-based (1 = first position).

  Returns {:ok, device} on success, {:error, changeset} on failure.
  """
  def reorder_device(device_id, new_position) when is_integer(new_position) and new_position > 0 do
    Repo.transaction(fn ->
      device = Repo.get!(DeviceSchema, device_id)
      site_id = device.site_id

      # Get all devices in site excluding the one being moved, in current order
      other_devices =
        Repo.all(
          from(d in DeviceSchema,
            where: d.site_id == ^site_id and d.id != ^device_id,
            order_by: [asc: d.display_order, asc: d.name]
          )
        )

      # Insert at new position (1-based index, convert to 0-based for List.insert_at)
      new_order = List.insert_at(other_devices, new_position - 1, device)

      {device_ids, orders} =
        new_order
        |> Enum.with_index(1)
        |> Enum.map(fn {d, order} -> {d.id, order} end)
        |> Enum.unzip()

      device_ids = Enum.map(device_ids, &Ecto.UUID.dump!/1)
      now = DateTime.truncate(DateTime.utc_now(), :second)

      query =
        from(d in DeviceSchema,
          join:
            data in fragment(
              "SELECT * FROM unnest(?::uuid[], ?::int[]) AS data(id, ordering)",
              ^device_ids,
              ^orders
            ),
          on: d.id == data.id,
          update: [set: [display_order: field(data, :ordering), updated_at: ^now]]
        )

      Repo.update_all(query, [])

      Repo.get!(DeviceSchema, device_id)
    end)
  end

  @doc """
  Reset all devices in organization to alphabetical order.

  Clears display_order field for all devices, allowing them to fall back to alphabetical sorting.
  """
  def reset_organization_device_order(organization_id) do
    Repo.update_all(
      from(d in DeviceSchema,
        join: s in assoc(d, :site),
        where: s.organization_id == ^organization_id
      ),
      set: [display_order: nil, updated_at: DateTime.truncate(DateTime.utc_now(), :second)]
    )
  end

  defp broadcast_device_change(organization_id, event) do
    Phoenix.PubSub.broadcast(
      Towerops.PubSub,
      "devices:org:#{organization_id}",
      {event, organization_id}
    )
  end
end