# Copyright (c) 2024 Ferrosoft GmbH. All rights reserved.
import logging
from dataclasses import dataclass
from datetime import timedelta
from urllib.parse import urljoin
from uuid import UUID
from asgiref.sync import sync_to_async
from django.core.mail import EmailMultiAlternatives, mail_managers
from django.db import IntegrityError, transaction
from django.template.loader import render_to_string
from django.utils import translation
from django.utils.translation import gettext as _
from uuid_extensions import uuid7
from ferrosoft.apps.date_time import now_utc
from ferrosoft.apps.ferrobase.auth.roles import Role
from ferrosoft.apps.ferrobase.models import AccountActivationRequest, User, Organization
from ferrosoft.apps.ferrobase.models.registry import ModelRegistry
from ferrosoft.apps.ferrobase.services.tenant_mgm import (
add_tenant,
TenantOptions,
STANDARD_FIXTURES,
)
from ferrosoft.apps.ferrobase.services.user_mgm import add_user, UserOptions
[docs]
@dataclass
class SignupRequest:
"""Inbound signup payload captured from the public registration form."""
email: str
company: str
first_name: str
last_name: str
language: str
[docs]
@sync_to_async
def initiate_signup(request: SignupRequest, frontend_uri: str) -> bool:
"""Create tenant + organisation and send the account-activation email.
The tenant is created inactive with a two-day expiry; if the user fails
to follow the activation link in time the cleanup service tears it down.
The first user is created later by :func:`activate_account`.
Args:
request: Sanitised signup payload.
frontend_uri: Absolute base URL of the frontend, used to build the
activation link embedded in the welcome email.
Returns:
bool: ``True`` on success, ``False`` when the request is silently
refused (existing user, mail-server unreachable, race-condition
``IntegrityError``).
"""
from django.conf import settings
if request.language:
translation.activate(request.language)
tenant_id = uuid7()
# Set early expiry date, to delete tenant resources if the user does not
# finish activation.
expiry = now_utc() + timedelta(days=2)
existing_user = User.objects.filter(username=request.email).first()
if existing_user is not None:
logging.info(
"Refusing to create account for existing user %s", str(existing_user)
)
return False
try:
organization = Organization.objects.create(
name=request.company,
contact_email=request.email,
contact_first_name=request.first_name,
contact_last_name=request.last_name,
subscription=ModelRegistry.SubscriptionPlan.get(
UUID("01999994-6610-7964-912b-cd0fe8a69c73")
),
)
tenant = add_tenant(
TenantOptions(
id=str(tenant_id),
name=request.company,
# Create inactive, because user will have to finish the activation
# by clicking a link in the welcome mail. Only then is the tenant enabled.
active=False,
expiry_date=expiry,
language=request.language,
fixtures=STANDARD_FIXTURES,
organization=organization,
)
)
activation_req = AccountActivationRequest.create(tenant, expiry)
logging.info("Created %s", str(activation_req))
template_ctx = {
"tenant": tenant,
"activation_url": urljoin(frontend_uri, activation_req.get_absolute_url()),
}
welcome_content_html = render_to_string(
"ferrobase/onboarding/mail/activation.html",
context=template_ctx,
)
welcome_content_text = render_to_string(
"ferrobase/onboarding/mail/activation.txt",
context=template_ctx,
)
msg = EmailMultiAlternatives(
_("Welcome to Emiflow"),
welcome_content_text,
settings.EMIFLOW.get(
"SUPPORT_ADDRESS",
getattr(settings, "DEFAULT_FROM_EMAIL"),
),
[organization.contact_email],
)
msg.attach_alternative(welcome_content_html, "text/html")
msg.send()
mail_managers(
"Emiflow Account Created",
render_to_string(
"ferrobase/onboarding/mail/manager.txt",
context={
"request": request,
"tenant": tenant,
"activation": activation_req,
},
),
)
return True
except ConnectionRefusedError:
logging.exception("Failed mailing account activation email")
return False
except IntegrityError as e:
logging.info(
"Refusing to activate account for %s: tenant already exists (detail: %s)",
request.email,
str(e),
)
return False
[docs]
def activate_account(activation: AccountActivationRequest, user_password: str):
"""Complete onboarding by creating the first user and activating the tenant.
The user is granted the broad ``EMIFLOW_USER``/``EMIFLOW_MANAGER``/
``USER_MANAGER`` roles so the new organisation has at least one full
administrator. Everything runs inside a single atomic transaction; if
any step raises the tenant remains inactive and the activation request
is preserved for retry.
Args:
activation: The activation record discovered via the email link.
user_password: Cleartext password chosen by the user — hashed by
Django's user-creation machinery.
"""
with transaction.atomic():
tenant = activation.tenant
org = tenant.organization
add_user(
UserOptions(
username=org.contact_email,
email=org.contact_email,
first_name=org.contact_first_name,
last_name=org.contact_last_name,
organization=org,
password=user_password,
# Create user with wide-ranging roles, as it is the
# first user in the organization.
roles=[
Role.EMIFLOW_USER,
Role.EMIFLOW_MANAGER,
Role.USER_MANAGER,
],
)
)
tenant.active = True
tenant.save(update_fields=["active"])
activation.delete()