Graham McIntire
76545ee31c
- Add handle_info for {:alert_changed, org_id} in AlertLive.Index and DashboardLive
to properly handle organization-level alert change broadcasts
- Fix get_primary_owner query to prioritize is_default memberships first,
ensuring the organization creator is always returned when multiple owners exist
Resolves 5 test failures related to PubSub message handling and
deterministic primary owner selection.
572 lines
17 KiB
Elixir
572 lines
17 KiB
Elixir
defmodule Towerops.Organizations do
|
|
@moduledoc """
|
|
The Organizations context.
|
|
"""
|
|
|
|
import Ecto.Query
|
|
|
|
alias Towerops.Accounts.User
|
|
alias Towerops.Agents.AgentAssignment
|
|
alias Towerops.Contexts.ConfigChangeTracker
|
|
alias Towerops.Devices.Device
|
|
alias Towerops.Organizations.Invitation
|
|
alias Towerops.Organizations.Membership
|
|
alias Towerops.Organizations.Organization
|
|
alias Towerops.Organizations.Policy
|
|
alias Towerops.Organizations.SubscriptionLimits
|
|
alias Towerops.Repo
|
|
alias Towerops.Sites.Site
|
|
|
|
## Organizations
|
|
|
|
@doc """
|
|
Returns the list of organizations for a user.
|
|
"""
|
|
@spec list_user_organizations(String.t()) :: [Organization.t()]
|
|
def list_user_organizations(user_id) do
|
|
Repo.all(
|
|
from(o in Organization,
|
|
join: m in Membership,
|
|
on: m.organization_id == o.id,
|
|
where: m.user_id == ^user_id,
|
|
order_by: [desc: m.inserted_at],
|
|
preload: [memberships: m]
|
|
)
|
|
)
|
|
end
|
|
|
|
@doc """
|
|
Checks if a user has access to an organization.
|
|
"""
|
|
@spec user_has_access?(String.t(), String.t()) :: boolean()
|
|
def user_has_access?(user_id, organization_id) do
|
|
Repo.exists?(
|
|
from m in Membership,
|
|
where: m.user_id == ^user_id and m.organization_id == ^organization_id
|
|
)
|
|
end
|
|
|
|
@doc """
|
|
Gets a single organization by ID.
|
|
"""
|
|
@spec get_organization!(String.t()) :: Organization.t()
|
|
def get_organization!(id), do: Repo.get!(Organization, id)
|
|
|
|
@doc """
|
|
Gets a single organization by slug.
|
|
"""
|
|
@spec get_organization_by_slug!(String.t()) :: Organization.t()
|
|
def get_organization_by_slug!(slug) do
|
|
Repo.get_by!(Organization, slug: slug)
|
|
end
|
|
|
|
@doc """
|
|
Creates an organization and adds the creator as owner.
|
|
|
|
## Options
|
|
* `:bypass_limits` - When true, bypasses free organization limit (for superusers)
|
|
"""
|
|
@dialyzer {:nowarn_function, create_organization: 2}
|
|
@dialyzer {:nowarn_function, create_organization: 3}
|
|
@spec create_organization(map(), String.t(), Keyword.t()) :: {:ok, Organization.t()} | {:error, Ecto.Changeset.t()}
|
|
def create_organization(attrs, user_id, opts \\ []) do
|
|
bypass_limits = Keyword.get(opts, :bypass_limits, false)
|
|
subscription_plan = Map.get(attrs, :subscription_plan) || Map.get(attrs, "subscription_plan") || "free"
|
|
|
|
# Check free org limit before creating (unless bypassing or creating non-free org)
|
|
with :ok <- check_free_org_limit(bypass_limits, subscription_plan, user_id, attrs) do
|
|
do_create_organization(attrs, user_id)
|
|
end
|
|
end
|
|
|
|
defp check_free_org_limit(true, _plan, _user_id, _attrs), do: :ok
|
|
defp check_free_org_limit(_bypass, plan, _user_id, _attrs) when plan != "free", do: :ok
|
|
|
|
defp check_free_org_limit(false, "free", user_id, attrs) do
|
|
if SubscriptionLimits.can_create_free_organization?(user_id) do
|
|
:ok
|
|
else
|
|
changeset =
|
|
%Organization{}
|
|
|> Organization.changeset(attrs)
|
|
|> Ecto.Changeset.add_error(
|
|
:base,
|
|
"You already have a free organization. Upgrade your existing organization to create additional ones."
|
|
)
|
|
|
|
{:error, changeset}
|
|
end
|
|
end
|
|
|
|
defp do_create_organization(attrs, user_id) do
|
|
# Check if user has any existing organizations
|
|
has_existing_orgs =
|
|
Repo.exists?(
|
|
from m in Membership,
|
|
where: m.user_id == ^user_id
|
|
)
|
|
|
|
multi =
|
|
Ecto.Multi.new()
|
|
|> Ecto.Multi.insert(:organization, Organization.changeset(%Organization{}, attrs))
|
|
|> Ecto.Multi.insert(:membership, fn %{organization: organization} ->
|
|
Membership.changeset(%Membership{}, %{
|
|
organization_id: organization.id,
|
|
user_id: user_id,
|
|
role: :owner,
|
|
# Set as default if this is the user's first organization
|
|
is_default: !has_existing_orgs
|
|
})
|
|
end)
|
|
|
|
case Repo.transaction(multi) do
|
|
{:ok, %{organization: organization}} -> {:ok, organization}
|
|
{:error, :organization, changeset, _} -> {:error, changeset}
|
|
{:error, :membership, changeset, _} -> {:error, changeset}
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Updates an organization.
|
|
"""
|
|
def update_organization(%Organization{} = organization, attrs) do
|
|
old_config = ConfigChangeTracker.capture_config_state(organization)
|
|
old_use_sites = organization.use_sites
|
|
|
|
with {:ok, updated_organization} <- organization |> Organization.changeset(attrs) |> Repo.update() do
|
|
propagate_organization_changes(updated_organization, old_config, old_use_sites)
|
|
{:ok, updated_organization}
|
|
end
|
|
end
|
|
|
|
defp propagate_organization_changes(updated_organization, old_config, old_use_sites) do
|
|
propagate_snmp_changes(updated_organization, old_config)
|
|
propagate_mikrotik_changes(updated_organization, old_config)
|
|
handle_sites_disabled(updated_organization, old_use_sites)
|
|
end
|
|
|
|
defp propagate_snmp_changes(updated_organization, old_config) do
|
|
if ConfigChangeTracker.snmp_changed?(updated_organization, old_config) do
|
|
Towerops.Devices.propagate_organization_community_change(
|
|
updated_organization.id,
|
|
updated_organization.snmp_community
|
|
)
|
|
end
|
|
end
|
|
|
|
defp propagate_mikrotik_changes(updated_organization, old_config) do
|
|
if ConfigChangeTracker.mikrotik_changed?(updated_organization, old_config) do
|
|
mikrotik_attrs = ConfigChangeTracker.extract_mikrotik_attrs(updated_organization)
|
|
|
|
Towerops.Devices.propagate_organization_mikrotik_change(
|
|
updated_organization.id,
|
|
mikrotik_attrs
|
|
)
|
|
end
|
|
end
|
|
|
|
defp handle_sites_disabled(updated_organization, old_use_sites) do
|
|
if old_use_sites && !updated_organization.use_sites do
|
|
clear_all_site_assignments(updated_organization.id)
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Deletes an organization.
|
|
"""
|
|
def delete_organization(%Organization{} = organization) do
|
|
Repo.delete(organization)
|
|
end
|
|
|
|
@doc """
|
|
Returns an `%Ecto.Changeset{}` for tracking organization changes.
|
|
"""
|
|
def change_organization(%Organization{} = organization, attrs \\ %{}) do
|
|
Organization.changeset(organization, attrs)
|
|
end
|
|
|
|
## Memberships
|
|
|
|
@doc """
|
|
Gets a user's membership for an organization.
|
|
"""
|
|
def get_membership(organization_id, user_id) do
|
|
Repo.get_by(Membership, organization_id: organization_id, user_id: user_id)
|
|
end
|
|
|
|
@doc """
|
|
Gets a user's membership for an organization, raises if not found.
|
|
"""
|
|
def get_membership!(organization_id, user_id) do
|
|
Repo.get_by!(Membership, organization_id: organization_id, user_id: user_id)
|
|
end
|
|
|
|
@doc """
|
|
Lists all memberships for an organization.
|
|
"""
|
|
def list_organization_memberships(organization_id) do
|
|
Repo.all(
|
|
from(m in Membership,
|
|
where: m.organization_id == ^organization_id,
|
|
preload: [:user],
|
|
order_by: [asc: m.inserted_at]
|
|
)
|
|
)
|
|
end
|
|
|
|
@doc """
|
|
Lists all members for an organization with preloaded users.
|
|
Ordered by role priority (owner first) then email.
|
|
"""
|
|
def list_organization_members(organization_id) do
|
|
Repo.all(
|
|
from(m in Membership,
|
|
where: m.organization_id == ^organization_id,
|
|
join: u in assoc(m, :user),
|
|
preload: [user: u],
|
|
order_by: [
|
|
fragment(
|
|
"CASE ? WHEN 'owner' THEN 0 WHEN 'admin' THEN 1 WHEN 'member' THEN 2 WHEN 'viewer' THEN 3 END",
|
|
m.role
|
|
),
|
|
asc: u.email
|
|
]
|
|
)
|
|
)
|
|
end
|
|
|
|
@doc """
|
|
Removes a member from an organization. Owners cannot be removed.
|
|
"""
|
|
def remove_member(organization_id, user_id) do
|
|
case get_membership(organization_id, user_id) do
|
|
%Membership{role: :owner} -> {:error, :cannot_remove_owner}
|
|
%Membership{} = membership -> delete_membership(membership)
|
|
nil -> {:error, :not_found}
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Updates a member's role. Owners cannot have their role changed.
|
|
"""
|
|
def update_member_role(organization_id, user_id, new_role) do
|
|
case get_membership(organization_id, user_id) do
|
|
%Membership{role: :owner} -> {:error, :cannot_change_owner_role}
|
|
%Membership{} = membership -> update_membership(membership, %{role: new_role})
|
|
nil -> {:error, :not_found}
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Lists users who should receive alert notifications for an organization.
|
|
Returns owners and admins.
|
|
"""
|
|
def list_organization_notification_recipients(organization_id) do
|
|
Repo.all(
|
|
from(u in User,
|
|
join: m in Membership,
|
|
on: m.user_id == u.id,
|
|
where: m.organization_id == ^organization_id,
|
|
where: m.role in [:owner, :admin],
|
|
select: u
|
|
)
|
|
)
|
|
end
|
|
|
|
@doc """
|
|
Creates a membership (adds a user to an organization).
|
|
"""
|
|
def create_membership(attrs) do
|
|
%Membership{}
|
|
|> Membership.changeset(attrs)
|
|
|> Repo.insert()
|
|
end
|
|
|
|
@doc """
|
|
Updates a membership (changes a user's role).
|
|
"""
|
|
def update_membership(%Membership{} = membership, attrs) do
|
|
membership
|
|
|> Membership.changeset(attrs)
|
|
|> Repo.update()
|
|
end
|
|
|
|
@doc """
|
|
Deletes a membership (removes a user from an organization).
|
|
"""
|
|
def delete_membership(%Membership{} = membership) do
|
|
Repo.delete(membership)
|
|
end
|
|
|
|
@doc """
|
|
Gets the user's default organization membership.
|
|
"""
|
|
def get_default_membership(user_id) do
|
|
Repo.one(
|
|
from m in Membership,
|
|
where: m.user_id == ^user_id and m.is_default == true,
|
|
preload: [:organization]
|
|
)
|
|
end
|
|
|
|
@doc """
|
|
Sets an organization as the user's default.
|
|
Clears any existing default and sets the new one.
|
|
"""
|
|
def set_default_organization(user_id, organization_id) do
|
|
# Verify user has access to this organization
|
|
if user_has_access?(user_id, organization_id) do
|
|
multi =
|
|
Ecto.Multi.new()
|
|
|> Ecto.Multi.update_all(:clear_old_default, clear_default_query(user_id), set: [is_default: false])
|
|
|> Ecto.Multi.update_all(
|
|
:set_new_default,
|
|
set_default_query(user_id, organization_id),
|
|
set: [is_default: true]
|
|
)
|
|
|
|
case Repo.transaction(multi) do
|
|
{:ok, _} -> {:ok, get_organization!(organization_id)}
|
|
{:error, _, changeset, _} -> {:error, changeset}
|
|
end
|
|
else
|
|
{:error, :not_found}
|
|
end
|
|
end
|
|
|
|
defp clear_default_query(user_id) do
|
|
from m in Membership,
|
|
where: m.user_id == ^user_id and m.is_default == true
|
|
end
|
|
|
|
defp set_default_query(user_id, organization_id) do
|
|
from m in Membership,
|
|
where: m.user_id == ^user_id and m.organization_id == ^organization_id
|
|
end
|
|
|
|
## Invitations
|
|
|
|
@doc """
|
|
Lists pending invitations for an organization.
|
|
"""
|
|
def list_pending_invitations(organization_id) do
|
|
now = DateTime.utc_now()
|
|
|
|
Repo.all(
|
|
from(i in Invitation,
|
|
where: i.organization_id == ^organization_id,
|
|
where: is_nil(i.accepted_at),
|
|
where: i.expires_at > ^now,
|
|
preload: [:invited_by],
|
|
order_by: [desc: i.inserted_at]
|
|
)
|
|
)
|
|
end
|
|
|
|
@doc """
|
|
Creates an invitation.
|
|
"""
|
|
def create_invitation(attrs) do
|
|
%Invitation{}
|
|
|> Invitation.changeset(attrs)
|
|
|> Repo.insert()
|
|
end
|
|
|
|
@doc """
|
|
Gets an invitation by token.
|
|
"""
|
|
def get_invitation_by_token(token) do
|
|
now = DateTime.utc_now()
|
|
|
|
Repo.one(
|
|
from(i in Invitation,
|
|
where: i.token == ^token,
|
|
where: is_nil(i.accepted_at),
|
|
where: i.expires_at > ^now,
|
|
preload: [:organization]
|
|
)
|
|
)
|
|
end
|
|
|
|
@doc """
|
|
Accepts an invitation and creates a membership.
|
|
"""
|
|
@dialyzer {:nowarn_function, accept_invitation: 2}
|
|
def accept_invitation(invitation, user_id) do
|
|
multi = Ecto.Multi.new()
|
|
|
|
multi =
|
|
Ecto.Multi.update(multi, :invitation, fn _ ->
|
|
Ecto.Changeset.change(invitation, %{
|
|
accepted_at: DateTime.truncate(DateTime.utc_now(), :second),
|
|
accepted_by_id: user_id
|
|
})
|
|
end)
|
|
|
|
multi =
|
|
Ecto.Multi.insert(multi, :membership, fn _ ->
|
|
Membership.changeset(%Membership{}, %{
|
|
organization_id: invitation.organization_id,
|
|
user_id: user_id,
|
|
role: invitation.role
|
|
})
|
|
end)
|
|
|
|
case Repo.transaction(multi) do
|
|
{:ok, %{membership: membership}} -> {:ok, membership}
|
|
{:error, _failed_operation, changeset, _changes_so_far} -> {:error, changeset}
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Deletes an invitation.
|
|
"""
|
|
def delete_invitation(%Invitation{} = invitation) do
|
|
Repo.delete(invitation)
|
|
end
|
|
|
|
## Bulk Configuration Updates
|
|
|
|
@doc """
|
|
Applies organization's SNMP configuration to all devices in the organization.
|
|
|
|
This updates all devices records to use the organization's SNMP version and community string,
|
|
overwriting any existing device-level or site-level configurations.
|
|
|
|
Returns the number of device records updated.
|
|
"""
|
|
def apply_snmp_config_to_all_equipment(organization_id) do
|
|
organization = get_organization!(organization_id)
|
|
|
|
Repo.update_all(from(e in Device, join: s in assoc(e, :site), where: s.organization_id == ^organization_id),
|
|
set: [
|
|
snmp_version: organization.snmp_version,
|
|
snmp_community: organization.snmp_community,
|
|
updated_at: DateTime.truncate(DateTime.utc_now(), :second)
|
|
]
|
|
)
|
|
end
|
|
|
|
@doc """
|
|
Applies organization's default agent to all devices in the organization.
|
|
|
|
This creates/updates agent assignments for all devices to use the organization's
|
|
default agent, overwriting any existing device-level or site-level assignments.
|
|
|
|
Returns the number of device records assigned.
|
|
"""
|
|
def apply_agent_to_all_equipment(organization_id) do
|
|
organization = Repo.get!(Organization, organization_id)
|
|
|
|
if organization.default_agent_token_id do
|
|
# device IDs for this organization (includes devices with and without sites)
|
|
device_ids =
|
|
Repo.all(from(e in Device, where: e.organization_id == ^organization_id, select: e.id))
|
|
|
|
# Delete existing assignments
|
|
Repo.delete_all(from(a in AgentAssignment, where: a.device_id in ^device_ids))
|
|
|
|
# Create new assignments
|
|
now = DateTime.truncate(DateTime.utc_now(), :second)
|
|
|
|
assignments =
|
|
Enum.map(device_ids, fn device_id ->
|
|
%{
|
|
device_id: device_id,
|
|
agent_token_id: organization.default_agent_token_id,
|
|
inserted_at: now,
|
|
updated_at: now
|
|
}
|
|
end)
|
|
|
|
{count, _} = Repo.insert_all(AgentAssignment, assignments)
|
|
{count, nil}
|
|
else
|
|
# No default agent set, delete all assignments
|
|
device_ids =
|
|
Repo.all(from(e in Device, where: e.organization_id == ^organization_id, select: e.id))
|
|
|
|
Repo.delete_all(from(a in AgentAssignment, where: a.device_id in ^device_ids))
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Clears all site_id assignments for devices in an organization when sites are disabled.
|
|
|
|
Returns the number of device records updated.
|
|
"""
|
|
def clear_all_site_assignments(organization_id) do
|
|
# Clear site assignments from all devices
|
|
Repo.update_all(
|
|
from(d in Device, where: d.organization_id == ^organization_id),
|
|
set: [
|
|
site_id: nil,
|
|
updated_at: DateTime.truncate(DateTime.utc_now(), :second)
|
|
]
|
|
)
|
|
|
|
# Delete all sites for the organization
|
|
Repo.delete_all(from(s in Site, where: s.organization_id == ^organization_id))
|
|
end
|
|
|
|
## Authorization
|
|
|
|
@doc """
|
|
Checks if a membership has permission to perform an action on a resource.
|
|
|
|
## Examples
|
|
|
|
iex> can?(membership, :edit, :device)
|
|
true
|
|
|
|
iex> can?(membership, :delete, :organization)
|
|
false
|
|
"""
|
|
defdelegate can?(membership, action, resource), to: Policy
|
|
|
|
## Billing
|
|
|
|
@doc """
|
|
Gets an organization by Stripe customer ID.
|
|
|
|
Returns nil if not found.
|
|
"""
|
|
@spec get_organization_by_stripe_customer_id(String.t()) :: Organization.t() | nil
|
|
def get_organization_by_stripe_customer_id(stripe_customer_id) do
|
|
Repo.get_by(Organization, stripe_customer_id: stripe_customer_id)
|
|
end
|
|
|
|
@doc """
|
|
Lists all organizations with active paid subscriptions.
|
|
|
|
Returns organizations where subscription_status is "active" or "trialing".
|
|
Used by BillingSyncWorker to sync usage to Stripe.
|
|
"""
|
|
@spec list_organizations_with_active_subscriptions() :: [Organization.t()]
|
|
def list_organizations_with_active_subscriptions do
|
|
Repo.all(
|
|
from o in Organization,
|
|
where: o.subscription_status in ["active", "trialing"],
|
|
where: not is_nil(o.stripe_customer_id)
|
|
)
|
|
end
|
|
|
|
@doc """
|
|
Gets the primary owner of an organization.
|
|
|
|
Returns the user with the :owner role. If multiple owners exist (shouldn't happen),
|
|
returns the first one by insertion date.
|
|
"""
|
|
@spec get_primary_owner(Organization.t()) :: User.t()
|
|
def get_primary_owner(%Organization{} = organization) do
|
|
Repo.one!(
|
|
from m in Membership,
|
|
join: u in assoc(m, :user),
|
|
where: m.organization_id == ^organization.id,
|
|
where: m.role == :owner,
|
|
order_by: [desc: m.is_default, asc: m.inserted_at, asc: m.id],
|
|
limit: 1,
|
|
select: u
|
|
)
|
|
end
|
|
end
|