139 lines
3.7 KiB
Elixir
139 lines
3.7 KiB
Elixir
defmodule Towerops.Devices.Firmware do
|
|
@moduledoc """
|
|
Context for managing firmware releases and device firmware history.
|
|
"""
|
|
|
|
import Ecto.Query, warn: false
|
|
|
|
alias Towerops.Devices.DeviceFirmwareHistory
|
|
alias Towerops.Devices.FirmwareRelease
|
|
alias Towerops.Repo
|
|
|
|
require Logger
|
|
|
|
@doc """
|
|
Creates or updates a firmware release record.
|
|
|
|
Uses vendor + product_line as the unique key. If a record exists, it will be updated.
|
|
Otherwise, a new record is created.
|
|
|
|
## Examples
|
|
|
|
iex> upsert_firmware_release(%{vendor: "mikrotik", product_line: "routeros", version: "7.14.1", fetched_at: DateTime.utc_now()})
|
|
{:ok, %FirmwareRelease{}}
|
|
|
|
iex> upsert_firmware_release(%{vendor: "mikrotik"})
|
|
{:error, %Ecto.Changeset{}}
|
|
"""
|
|
def upsert_firmware_release(attrs) do
|
|
vendor = Map.get(attrs, :vendor) || Map.get(attrs, "vendor")
|
|
product_line = Map.get(attrs, :product_line) || Map.get(attrs, "product_line")
|
|
|
|
# Find existing record (handle nil product_line)
|
|
existing =
|
|
if vendor && product_line do
|
|
Repo.one(from(f in FirmwareRelease, where: f.vendor == ^vendor and f.product_line == ^product_line))
|
|
end
|
|
|
|
changeset =
|
|
case existing do
|
|
nil -> FirmwareRelease.changeset(%FirmwareRelease{}, attrs)
|
|
record -> FirmwareRelease.changeset(record, attrs)
|
|
end
|
|
|
|
Repo.insert_or_update(changeset)
|
|
end
|
|
|
|
@doc """
|
|
Gets the latest firmware release for a vendor and product line.
|
|
|
|
Returns nil if either vendor or product_line is nil.
|
|
|
|
## Examples
|
|
|
|
iex> get_latest_firmware_release("mikrotik", "routeros")
|
|
%FirmwareRelease{}
|
|
|
|
iex> get_latest_firmware_release("unknown", "unknown")
|
|
nil
|
|
|
|
iex> get_latest_firmware_release("mikrotik", nil)
|
|
nil
|
|
"""
|
|
def get_latest_firmware_release(nil, _product_line), do: nil
|
|
def get_latest_firmware_release(_vendor, nil), do: nil
|
|
|
|
def get_latest_firmware_release(vendor, product_line) do
|
|
Repo.one(from(f in FirmwareRelease, where: f.vendor == ^vendor and f.product_line == ^product_line))
|
|
end
|
|
|
|
@doc """
|
|
Lists all firmware releases.
|
|
|
|
## Examples
|
|
|
|
iex> list_firmware_releases()
|
|
[%FirmwareRelease{}, ...]
|
|
"""
|
|
def list_firmware_releases do
|
|
Repo.all(FirmwareRelease)
|
|
end
|
|
|
|
@doc """
|
|
Lists firmware history for a device, most recent first.
|
|
|
|
## Examples
|
|
|
|
iex> list_device_firmware_history(device_id, 20)
|
|
[%DeviceFirmwareHistory{}, ...]
|
|
"""
|
|
def list_device_firmware_history(snmp_device_id, limit \\ 20) do
|
|
DeviceFirmwareHistory
|
|
|> where([h], h.snmp_device_id == ^snmp_device_id)
|
|
|> order_by([h], desc: h.detected_at)
|
|
|> limit(^limit)
|
|
|> Repo.all()
|
|
end
|
|
|
|
@doc """
|
|
Logs a firmware version change for a device.
|
|
|
|
Creates a firmware history record and broadcasts the change via PubSub.
|
|
|
|
## Examples
|
|
|
|
iex> log_firmware_change(snmp_device_id, "7.12.0", "7.14.1")
|
|
{:ok, %DeviceFirmwareHistory{}}
|
|
"""
|
|
def log_firmware_change(snmp_device_id, old_version, new_version) do
|
|
attrs = %{
|
|
snmp_device_id: snmp_device_id,
|
|
old_version: old_version,
|
|
new_version: new_version,
|
|
detected_at: DateTime.utc_now(),
|
|
detection_method: "discovery"
|
|
}
|
|
|
|
changeset = DeviceFirmwareHistory.changeset(%DeviceFirmwareHistory{}, attrs)
|
|
|
|
case Repo.insert(changeset) do
|
|
{:ok, _history} = result ->
|
|
Logger.info("Firmware version changed",
|
|
snmp_device_id: snmp_device_id,
|
|
old_version: old_version,
|
|
new_version: new_version
|
|
)
|
|
|
|
Phoenix.PubSub.broadcast(
|
|
Towerops.PubSub,
|
|
"device:#{snmp_device_id}",
|
|
{:firmware_changed, snmp_device_id, old_version, new_version}
|
|
)
|
|
|
|
result
|
|
|
|
error ->
|
|
error
|
|
end
|
|
end
|
|
end
|