poduck/smoothschedule
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
89fa8f81afc127288241bc6bc3041da09d54394c
smoothschedule/smoothschedule/smoothschedule/identity/core/quota_service.py
T
poduck 89fa8f81af Fix: Display correct SmoothSchedule logo in email preview 
Replaced the blank base64 encoded logo with the actual SmoothSchedule logo in the email rendering pipeline.

A Playwright E2E test was run to verify that the logo is correctly displayed in the email preview modal, ensuring it loads with natural dimensions and is visible.
2025-12-14 19:10:56 -05:00

647 lines
22 KiB
Python
Raw Blame History

"""
Quota Overage Service
Handles detection, tracking, and resolution of quota overages when tenants
exceed their plan limits (e.g., after downgrade or plan expiration).
Grace Period: 30 days
- Users can select which resources to archive
- After grace period, excess resources are auto-archived
- Archived resources become read-only (visible but not usable)
Email Notifications:
- Immediately when overage detected
- 7 days before grace period ends
- 1 day before grace period ends
"""
import logging
from datetime import timedelta
from django.utils import timezone
from django.db import transaction
from django.core.mail import send_mail
from django.template.loader import render_to_string
from django.conf import settings
from .models import Tenant, QuotaOverage
from smoothschedule.identity.users.models import User
logger = logging.getLogger(__name__)
class QuotaService:
"""
Service class for managing quota overages.
"""
GRACE_PERIOD_DAYS = 30
# Quota types and their corresponding models/counting logic
QUOTA_CONFIG = {
'MAX_ADDITIONAL_USERS': {
'model': 'smoothschedule.identity.users.models.User',
'display_name': 'additional team members',
'count_method': 'count_additional_users',
},
'MAX_RESOURCES': {
'model': 'schedule.models.Resource',
'display_name': 'resources',
'count_method': 'count_resources',
},
'MAX_SERVICES': {
'model': 'schedule.models.Service',
'display_name': 'services',
'count_method': 'count_services',
},
# Note: MAX_EMAIL_TEMPLATES quota removed - email templates are now system-wide
# using PuckEmailTemplate in the messaging app, not per-tenant
'MAX_AUTOMATED_TASKS': {
'model': 'schedule.models.ScheduledTask',
'display_name': 'automated tasks',
'count_method': 'count_automated_tasks',
},
}
def __init__(self, tenant: Tenant):
self.tenant = tenant
# =========================================================================
# Counting Methods
# =========================================================================
def count_additional_users(self) -> int:
"""Count additional users (excluding owner and archived)."""
return User.objects.filter(
tenant=self.tenant,
is_archived_by_quota=False
).exclude(
role=User.Role.TENANT_OWNER
).count()
def count_resources(self) -> int:
"""Count active resources (excluding archived)."""
from smoothschedule.scheduling.schedule.models import Resource
return Resource.objects.filter(is_archived_by_quota=False).count()
def count_services(self) -> int:
"""Count active services (excluding archived)."""
from smoothschedule.scheduling.schedule.models import Service
return Service.objects.filter(is_archived_by_quota=False).count()
# Note: count_email_templates removed - templates are now system-wide via PuckEmailTemplate
def count_automated_tasks(self) -> int:
"""Count automated tasks."""
from smoothschedule.scheduling.schedule.models import ScheduledTask
return ScheduledTask.objects.count()
# =========================================================================
# Limit Retrieval
# =========================================================================
def get_current_usage(self, quota_type: str) -> int:
"""Get the current usage for a quota type."""
config = self.QUOTA_CONFIG.get(quota_type)
if not config:
return 0
count_method = getattr(self, config['count_method'])
return count_method()
def get_limit(self, quota_type: str) -> int:
"""Get the current limit for a quota type based on tenant's billing plan."""
# Convert quota type to billing feature code
# e.g., MAX_ADDITIONAL_USERS -> max_users, MAX_RESOURCES -> max_resources
feature_code_map = {
'MAX_ADDITIONAL_USERS': 'max_users',
'MAX_RESOURCES': 'max_resources',
'MAX_SERVICES': 'max_services',
'MAX_AUTOMATED_TASKS': 'max_automated_tasks',
}
feature_code = feature_code_map.get(quota_type, quota_type.lower())
# Use billing entitlement system
limit = self.tenant.get_limit(feature_code)
# None means no limit defined = unlimited
if limit is None:
return -1 # -1 means unlimited
# 0 in billing system also means unlimited
if limit == 0:
return -1
return limit
# =========================================================================
# Overage Detection
# =========================================================================
def check_all_quotas(self) -> list[QuotaOverage]:
"""
Check all quota types for overages.
Returns list of newly created QuotaOverage records.
"""
new_overages = []
for quota_type, config in self.QUOTA_CONFIG.items():
overage = self.check_quota(quota_type)
if overage:
new_overages.append(overage)
return new_overages
def check_quota(self, quota_type: str) -> QuotaOverage | None:
"""
Check a specific quota type for overage.
Creates QuotaOverage record if over limit and none exists.
Returns the overage record or None.
"""
config = self.QUOTA_CONFIG.get(quota_type)
if not config:
logger.warning(f"Unknown quota type: {quota_type}")
return None
# Get current usage
count_method = getattr(self, config['count_method'])
current_usage = count_method()
# Get limit
limit = self.get_limit(quota_type)
# -1 means unlimited
if limit < 0:
return None
# Check if over limit
if current_usage <= limit:
# Not over limit - check if there's an active overage to resolve
self._resolve_overage_if_exists(quota_type)
return None
# Over limit - check for existing active overage
existing = QuotaOverage.objects.filter(
tenant=self.tenant,
quota_type=quota_type,
status='ACTIVE'
).first()
if existing:
# Update the existing overage with current counts
existing.current_usage = current_usage
existing.allowed_limit = limit
existing.save()
return existing
# Create new overage record
with transaction.atomic():
overage = QuotaOverage.objects.create(
tenant=self.tenant,
quota_type=quota_type,
current_usage=current_usage,
allowed_limit=limit,
overage_amount=current_usage - limit,
grace_period_days=self.GRACE_PERIOD_DAYS,
grace_period_ends_at=timezone.now() + timedelta(days=self.GRACE_PERIOD_DAYS)
)
# Send initial notification email
self.send_overage_notification(overage, 'initial')
logger.info(
f"Created quota overage for {self.tenant.name}: "
f"{quota_type} ({current_usage}/{limit})"
)
return overage
def _resolve_overage_if_exists(self, quota_type: str):
"""Resolve any existing active overage for this quota type."""
existing = QuotaOverage.objects.filter(
tenant=self.tenant,
quota_type=quota_type,
status='ACTIVE'
).first()
if existing:
existing.resolve('USER_DELETED')
logger.info(
f"Resolved quota overage for {self.tenant.name}: {quota_type}"
)
# =========================================================================
# Email Notifications
# =========================================================================
def send_overage_notification(self, overage: QuotaOverage, notification_type: str):
"""
Send email notification about quota overage.
notification_type:
- 'initial': First notification when overage detected
- 'week_reminder': 7 days before grace period ends
- 'day_reminder': 1 day before grace period ends
"""
# Get tenant owner
owner = User.objects.filter(
tenant=self.tenant,
role=User.Role.TENANT_OWNER
).first()
if not owner or not owner.email:
logger.warning(
f"Cannot send overage notification for {self.tenant.name}: no owner email"
)
return
config = self.QUOTA_CONFIG.get(overage.quota_type, {})
display_name = config.get('display_name', overage.quota_type)
# Prepare email context
context = {
'tenant': self.tenant,
'owner': owner,
'overage': overage,
'display_name': display_name,
'days_remaining': overage.days_remaining,
'grace_period_ends': overage.grace_period_ends_at,
'current_usage': overage.current_usage,
'allowed_limit': overage.allowed_limit,
'overage_amount': overage.overage_amount,
'manage_url': self._get_manage_url(),
'upgrade_url': self._get_upgrade_url(),
'export_url': self._get_export_url(),
}
# Select template based on notification type
if notification_type == 'initial':
subject = f"Action Required: Your {self.tenant.name} account has exceeded its quota"
template = 'emails/quota_overage_initial.html'
overage.initial_email_sent_at = timezone.now()
elif notification_type == 'week_reminder':
subject = f"Reminder: 7 days left to resolve quota overage for {self.tenant.name}"
template = 'emails/quota_overage_week_reminder.html'
overage.week_reminder_sent_at = timezone.now()
elif notification_type == 'day_reminder':
subject = f"Final Warning: 1 day left to resolve quota overage for {self.tenant.name}"
template = 'emails/quota_overage_day_reminder.html'
overage.day_reminder_sent_at = timezone.now()
else:
logger.error(f"Unknown notification type: {notification_type}")
return
overage.save()
# Render and send email
try:
html_message = render_to_string(template, context)
text_message = render_to_string(
template.replace('.html', '.txt'),
context
)
send_mail(
subject=subject,
message=text_message,
from_email=settings.DEFAULT_FROM_EMAIL,
recipient_list=[owner.email],
html_message=html_message,
fail_silently=False,
)
logger.info(
f"Sent {notification_type} overage email to {owner.email} "
f"for {self.tenant.name}"
)
except Exception as e:
logger.error(
f"Failed to send overage email to {owner.email}: {e}"
)
def _get_manage_url(self) -> str:
"""Get URL for quota management page."""
domain = self.tenant.get_primary_domain()
if domain:
return f"https://{domain.domain}/settings/quota"
return ""
def _get_upgrade_url(self) -> str:
"""Get URL for plan upgrade page."""
domain = self.tenant.get_primary_domain()
if domain:
return f"https://{domain.domain}/settings/subscription"
return ""
def _get_export_url(self) -> str:
"""Get URL for data export page."""
domain = self.tenant.get_primary_domain()
if domain:
return f"https://{domain.domain}/settings/export"
return ""
# =========================================================================
# Resource Archiving
# =========================================================================
def archive_resources(self, quota_type: str, resource_ids: list[int]) -> int:
"""
Archive specific resources selected by the user.
Returns the number of resources archived.
"""
count = 0
if quota_type == 'MAX_ADDITIONAL_USERS':
count = User.objects.filter(
tenant=self.tenant,
id__in=resource_ids,
is_archived_by_quota=False
).exclude(
role=User.Role.TENANT_OWNER # Never archive owner
).update(
is_archived_by_quota=True,
archived_by_quota_at=timezone.now()
)
elif quota_type == 'MAX_RESOURCES':
from smoothschedule.scheduling.schedule.models import Resource
count = Resource.objects.filter(
id__in=resource_ids,
is_archived_by_quota=False
).update(
is_archived_by_quota=True,
archived_by_quota_at=timezone.now()
)
elif quota_type == 'MAX_SERVICES':
from smoothschedule.scheduling.schedule.models import Service
count = Service.objects.filter(
id__in=resource_ids,
is_archived_by_quota=False
).update(
is_archived_by_quota=True,
archived_by_quota_at=timezone.now()
)
# Update overage record
overage = QuotaOverage.objects.filter(
tenant=self.tenant,
quota_type=quota_type,
status='ACTIVE'
).first()
if overage:
# Check if resolved
count_method = getattr(self, self.QUOTA_CONFIG[quota_type]['count_method'])
current_usage = count_method()
if current_usage <= overage.allowed_limit:
overage.resolve('USER_ARCHIVED', resource_ids)
return count
def unarchive_resource(self, quota_type: str, resource_id: int) -> bool:
"""
Unarchive a resource (swap with another that will be archived).
Returns True if successful.
"""
# Check if we have room to unarchive
count_method = getattr(self, self.QUOTA_CONFIG[quota_type]['count_method'])
current_usage = count_method()
limit = self.get_limit(quota_type)
if current_usage >= limit:
# No room - cannot unarchive without archiving another
return False
if quota_type == 'MAX_ADDITIONAL_USERS':
User.objects.filter(
id=resource_id,
tenant=self.tenant
).update(
is_archived_by_quota=False,
archived_by_quota_at=None
)
elif quota_type == 'MAX_RESOURCES':
from smoothschedule.scheduling.schedule.models import Resource
Resource.objects.filter(id=resource_id).update(
is_archived_by_quota=False,
archived_by_quota_at=None
)
elif quota_type == 'MAX_SERVICES':
from smoothschedule.scheduling.schedule.models import Service
Service.objects.filter(id=resource_id).update(
is_archived_by_quota=False,
archived_by_quota_at=None
)
return True
# =========================================================================
# Auto-Archive (Grace Period Expired)
# =========================================================================
def auto_archive_expired(self) -> dict:
"""
Auto-archive resources for overages where grace period has expired.
Archives the oldest/least recently used resources.
Returns dict with counts of archived resources by type.
"""
results = {}
expired_overages = QuotaOverage.objects.filter(
tenant=self.tenant,
status='ACTIVE',
grace_period_ends_at__lte=timezone.now()
)
for overage in expired_overages:
archived_ids = self._auto_archive_for_overage(overage)
if archived_ids:
overage.resolve('AUTO_ARCHIVED', archived_ids)
results[overage.quota_type] = len(archived_ids)
return results
def _auto_archive_for_overage(self, overage: QuotaOverage) -> list[int]:
"""
Auto-archive excess resources for a specific overage.
Archives the oldest resources first.
Returns list of archived resource IDs.
"""
quota_type = overage.quota_type
excess_count = overage.overage_amount
archived_ids = []
if quota_type == 'MAX_ADDITIONAL_USERS':
# Archive oldest non-owner users
users_to_archive = User.objects.filter(
tenant=self.tenant,
is_archived_by_quota=False
).exclude(
role=User.Role.TENANT_OWNER
).order_by('date_joined')[:excess_count]
for user in users_to_archive:
user.is_archived_by_quota = True
user.archived_by_quota_at = timezone.now()
user.save()
archived_ids.append(user.id)
elif quota_type == 'MAX_RESOURCES':
from smoothschedule.scheduling.schedule.models import Resource
resources = Resource.objects.filter(
is_archived_by_quota=False
).order_by('created_at')[:excess_count]
for resource in resources:
resource.is_archived_by_quota = True
resource.archived_by_quota_at = timezone.now()
resource.save()
archived_ids.append(resource.id)
elif quota_type == 'MAX_SERVICES':
from smoothschedule.scheduling.schedule.models import Service
services = Service.objects.filter(
is_archived_by_quota=False
).order_by('created_at')[:excess_count]
for service in services:
service.is_archived_by_quota = True
service.archived_by_quota_at = timezone.now()
service.save()
archived_ids.append(service.id)
return archived_ids
# =========================================================================
# Status Methods
# =========================================================================
def get_active_overages(self) -> list[dict]:
"""Get all active quota overages for this tenant."""
overages = QuotaOverage.objects.filter(
tenant=self.tenant,
status='ACTIVE'
)
return [
{
'id': o.id,
'quota_type': o.quota_type,
'display_name': self.QUOTA_CONFIG.get(o.quota_type, {}).get(
'display_name', o.quota_type
),
'current_usage': o.current_usage,
'allowed_limit': o.allowed_limit,
'overage_amount': o.overage_amount,
'days_remaining': o.days_remaining,
'grace_period_ends_at': o.grace_period_ends_at.isoformat() if o.grace_period_ends_at else None,
}
for o in overages
]
def has_active_overages(self) -> bool:
"""Check if tenant has any active quota overages."""
return QuotaOverage.objects.filter(
tenant=self.tenant,
status='ACTIVE'
).exists()
# =============================================================================
# Helper Functions
# =============================================================================
def check_tenant_quotas(tenant: Tenant) -> list[QuotaOverage]:
"""
Check all quotas for a tenant and create overage records if needed.
Call this after plan downgrades or billing failures.
"""
service = QuotaService(tenant)
return service.check_all_quotas()
def process_expired_grace_periods() -> dict:
"""
Process all tenants with expired grace periods.
Call this from a daily Celery task.
Returns:
dict with counts of processed overages and archived resources
"""
results = {
'overages_processed': 0,
'total_archived': 0,
}
# Find all tenants with expired overages
expired_overages = QuotaOverage.objects.filter(
status='ACTIVE',
grace_period_ends_at__lte=timezone.now()
).values_list('tenant_id', flat=True).distinct()
for tenant_id in expired_overages:
try:
tenant = Tenant.objects.get(id=tenant_id)
service = QuotaService(tenant)
archive_results = service.auto_archive_expired()
if archive_results:
logger.info(f"Auto-archived for {tenant.name}: {archive_results}")
results['overages_processed'] += len(archive_results)
results['total_archived'] += sum(archive_results.values())
except Tenant.DoesNotExist:
continue
except Exception as e:
logger.error(f"Error processing expired overages for tenant {tenant_id}: {e}")
return results
def send_grace_period_reminders() -> dict:
"""
Send reminder emails for overages approaching grace period end.
Call this from a daily Celery task.
Returns:
dict with counts of reminders sent
"""
now = timezone.now()
week_from_now = now + timedelta(days=7)
day_from_now = now + timedelta(days=1)
results = {
'week_reminders_sent': 0,
'day_reminders_sent': 0,
}
# 7-day reminders
week_overages = QuotaOverage.objects.filter(
status='ACTIVE',
week_reminder_sent_at__isnull=True,
grace_period_ends_at__lte=week_from_now,
grace_period_ends_at__gt=day_from_now
)
for overage in week_overages:
try:
service = QuotaService(overage.tenant)
service.send_overage_notification(overage, 'week_reminder')
results['week_reminders_sent'] += 1
except Exception as e:
logger.error(f"Error sending week reminder for overage {overage.id}: {e}")
# 1-day reminders
day_overages = QuotaOverage.objects.filter(
status='ACTIVE',
day_reminder_sent_at__isnull=True,
grace_period_ends_at__lte=day_from_now
)
for overage in day_overages:
try:
service = QuotaService(overage.tenant)
service.send_overage_notification(overage, 'day_reminder')
results['day_reminders_sent'] += 1
except Exception as e:
logger.error(f"Error sending day reminder for overage {overage.id}: {e}")
return results
Reference in New Issue View Git Blame Copy Permalink