Source code for ferrosoft.apps.ferrobase.services.cleanup

#  Copyright (c) 2024 Ferrosoft GmbH. All rights reserved.
import logging
from datetime import timedelta
from typing import List, Tuple

from django.apps import apps
from django.utils.translation import gettext as _

from ferrosoft.apps.date_time import now_utc
from ferrosoft.apps.ferrobase.models import Organization, StoredFile, Tenant
from ferrosoft.apps.ferrobase.tenant.context import active_tenant




class CleanupService:
    """Periodic database housekeeping orchestrator.

    Runs a list of cleanup tasks against the master database and against each
    active tenant database.  Tasks that raise are logged but do not abort the
    surrounding run, so unrelated cleanups continue to make progress.
    """

    DELETE_TENANT_DAYS = 7
    DELETE_EXAMINATION_REQUEST_DAYS = 14

    def __init__(self):
        """Resolve optional models and seed the master/tenant task lists.

        The ``befundung.ExaminationRequest`` model is optional: if the
        ``befundung`` app is not installed the corresponding tenant task
        becomes a no-op.
        """
        try:
            self.examination_request = apps.get_model("befundung", "ExaminationRequest")
        except LookupError:
            self.examination_request = None

        self.master_ops = [
            (
                "delete_expired_files",
                _("Deleted %d expired stored files."),
            ),
            (
                "deactivate_expired_tenants",
                _("Deactivated %d expired tenants."),
            ),
            (
                "delete_inactive_tenants",
                _("Deleted %d inactive tenants."),
            ),
            (
                "delete_empty_orgs",
                _("Deleted %d organizations without tenants."),
            ),
        ]

        self.tenant_ops = [
            (
                "delete_old_examination_requests",
                _("Deleted %d examination requests."),
            ),
        ]



    def cleanup_master_database(self):
        """Run cleanup tasks in master database, log results."""
        self._run_cleanup_tasks(self.master_ops)




    def cleanup_tenant_database(self, tenant: Tenant):
        """Run cleanup tasks in tenant database, log results."""
        with active_tenant(tenant):
            self._run_cleanup_tasks(self.tenant_ops)


    def _run_cleanup_tasks(self, tasks: List[Tuple[str, str]]):
        for method, message in tasks:
            try:
                logging.info(message % getattr(self, method)())
            except Exception as e:
                logging.error("Cleanup task %s failed", method)
                logging.exception(e)

    @staticmethod
    def _deletion_date(number_of_days):
        return now_utc() - timedelta(days=number_of_days)



    @staticmethod
    def delete_expired_files() -> int:
        """
        Deletes expired StoredFile objects and their associated files
        for all tenants.
        """
        total_deleted = 0

        for tenant in Tenant.objects.filter(active=True):
            with active_tenant(tenant):
                expired_files = StoredFile.objects.filter(expire_on__lt=now_utc())
                for stored_file in expired_files:
                    stored_file.delete()

                total_deleted += expired_files.count()

        return total_deleted




    @staticmethod
    def deactivate_expired_tenants() -> int:
        """Mark tenants past their ``expiry_date`` as inactive; return the count."""
        return Tenant.objects.filter(expiry_date__lt=now_utc(), active=True).update(
            active=False
        )




    def delete_inactive_tenants(self) -> int:
        """Delete tenants that have been inactive for at least :attr:`DELETE_TENANT_DAYS`.

        The two-phase pattern (first deactivate, then delete after a grace
        period) gives operators a window to reactivate a tenant before the
        record is gone.
        """
        deleted, _ = Tenant.objects.filter(
            expiry_date__lt=self._deletion_date(self.DELETE_TENANT_DAYS), active=False
        ).delete()
        return deleted




    def delete_empty_orgs(self) -> int:
        """Delete organisations that have no tenants left; return the count."""
        deleted, _ = Organization.objects.filter(tenants__isnull=True).delete()
        return deleted




    def delete_old_examination_requests(self) -> int:
        """Delete examination requests older than :attr:`DELETE_EXAMINATION_REQUEST_DAYS`.

        Returns ``0`` when the optional ``befundung`` app is not installed.
        """
        if not self.examination_request:
            return 0

        deleted, _ = self.examination_request.objects.filter(
            creation_date__lt=self._deletion_date(self.DELETE_EXAMINATION_REQUEST_DAYS)
        ).delete()
        return deleted