from __future__ import annotations from uuid import UUID from sqlalchemy.orm import Session from app.core.security import redacted_secret from app.core.time import as_utc, utc_now from app.domain.enums import ActivationStatus, DirectServerStatus, LicenceEventType, LicenceType from app.models.entities import Customer, IntegrationIdempotencyRecord, Licence from app.repositories.licensing import LicensingRepository from app.schemas.integrations import ( JoomlaActivationSummary, JoomlaCustomerRequest, JoomlaCustomerResponse, JoomlaLicenceDetailResponse, JoomlaLicenceListResponse, JoomlaLicenceSummary, JoomlaProvisionLicenceRequest, JoomlaProvisionLicenceResponse, JoomlaRemoteDeactivateResponse, ) from app.services.errors import INTEGRATION_RESOURCE_NOT_FOUND, INVALID_LICENCE from app.services.licensing import LicensingService from app.services.provisioning import ProvisioningService JOOMLA_SOURCE = "joomla" class JoomlaIntegrationService: def __init__(self, session: Session): self.session = session self.repository = LicensingRepository(session) self.licensing = LicensingService(session) def find_or_create_customer(self, request: JoomlaCustomerRequest) -> JoomlaCustomerResponse: customer = self._find_or_create_customer(request.joomla_user_id, request.email) self.session.commit() return self._customer_response(customer) def list_licences(self, joomla_user_id: str) -> JoomlaLicenceListResponse: customer = self._require_customer(joomla_user_id) licences = self.repository.list_customer_licences(customer) return JoomlaLicenceListResponse( customer=self._customer_response(customer), licences=[self._licence_summary(licence) for licence in licences], ) def licence_detail(self, joomla_user_id: str, licence_id: UUID) -> JoomlaLicenceDetailResponse: customer = self._require_customer(joomla_user_id) licence = self.repository.find_customer_licence(customer, licence_id) if licence is None: raise INTEGRATION_RESOURCE_NOT_FOUND activations = self.repository.list_activations_for_licence(licence) return JoomlaLicenceDetailResponse( licence=self._licence_summary(licence), activations=[self._activation_summary(activation) for activation in activations], ) def provision_licence(self, request: JoomlaProvisionLicenceRequest) -> JoomlaProvisionLicenceResponse: existing = self.repository.find_idempotency_record(JOOMLA_SOURCE, request.idempotency_key) if existing is not None and existing.licence_id is not None: licence = self.repository.get_licence_for_update(existing.licence_id) if licence is None: raise INVALID_LICENCE return JoomlaProvisionLicenceResponse( created=False, plaintextLicenceKey=None, licence=self._licence_summary(licence), ) customer = self._find_or_create_customer(request.joomla_user_id) provisioned = ProvisioningService(self.session).create_licence( licence_type=request.licence_type, activation_limit=request.activation_limit, subscription_expiry=request.subscription_expiry, update_entitlement_expiry=request.update_entitlement_expiry, support_entitlement_expiry=request.support_entitlement_expiry, customer_id=customer.id, ) licence = self.repository.get_licence_for_update(provisioned.licence_id) if licence is None: raise INVALID_LICENCE self.repository.add_idempotency_record( IntegrationIdempotencyRecord( source=JOOMLA_SOURCE, idempotency_key=request.idempotency_key, operation="provision_licence", licence_id=licence.id, response_json={ "licenceID": str(licence.id), "publicLicenceIdentifier": licence.public_licence_identifier, }, ) ) self.repository.add_event( event_type=LicenceEventType.LICENCE_CREATED, result="joomla_provisioned", licence=licence, detail={"source": JOOMLA_SOURCE, "customerID": str(customer.id)}, ) self.session.commit() return JoomlaProvisionLicenceResponse( created=True, plaintextLicenceKey=provisioned.licence_key, licence=self._licence_summary(licence), ) def remotely_deactivate( self, joomla_user_id: str, activation_id: UUID, ) -> JoomlaRemoteDeactivateResponse: customer = self._require_customer(joomla_user_id) activation = self.repository.find_activation_for_customer(customer, activation_id) if activation is None: raise INTEGRATION_RESOURCE_NOT_FOUND if activation.status == ActivationStatus.ACTIVE: activation.status = ActivationStatus.DEACTIVATED activation.deactivated_at = utc_now() self.repository.add_event( event_type=LicenceEventType.DEACTIVATION, result="joomla_remote_deactivation", licence=activation.licence, activation=activation, detail={"source": JOOMLA_SOURCE, "customerID": str(customer.id)}, ) self.session.commit() return JoomlaRemoteDeactivateResponse( ok=True, entitlementStatus=DirectServerStatus.ACTIVATION_REQUIRED, ) def _find_or_create_customer(self, joomla_user_id: str, email: str | None = None) -> Customer: customer = self.repository.find_customer_by_external_identity(JOOMLA_SOURCE, joomla_user_id) if customer is not None: if email is not None and customer.email != email: customer.email = email return customer customer = Customer( external_source=JOOMLA_SOURCE, external_customer_id=joomla_user_id, external_customer_reference=f"{JOOMLA_SOURCE}:{joomla_user_id}", email=email, ) self.repository.add_customer(customer) self.session.flush() self.repository.add_event( event_type=LicenceEventType.STATUS_CHANGE, result="joomla_customer_mapping_created", detail={"source": JOOMLA_SOURCE, "externalCustomerID": joomla_user_id}, ) return customer def _require_customer(self, joomla_user_id: str) -> Customer: customer = self.repository.find_customer_by_external_identity(JOOMLA_SOURCE, joomla_user_id) if customer is None: raise INVALID_LICENCE return customer def _customer_response(self, customer: Customer) -> JoomlaCustomerResponse: return JoomlaCustomerResponse( customerID=customer.id, externalSource=customer.external_source or JOOMLA_SOURCE, externalCustomerID=customer.external_customer_id or "", email=customer.email, ) def _licence_summary(self, licence: Licence) -> JoomlaLicenceSummary: active_activation_count = self.repository.count_active_activations(licence) return JoomlaLicenceSummary( licenceID=licence.id, publicLicenceIdentifier=licence.public_licence_identifier, redactedLicenceKey=redacted_secret(licence.licence_key_suffix), licenceType=licence.licence_type, status=licence.status, activationLimit=licence.activation_limit, activeActivationCount=active_activation_count, subscriptionExpiry=as_utc(licence.subscription_expiry) if licence.subscription_expiry else None, updateEntitlementExpiry=as_utc(licence.update_entitlement_expiry) if licence.update_entitlement_expiry else None, supportEntitlementExpiry=as_utc(licence.support_entitlement_expiry) if licence.support_entitlement_expiry else None, entitlements=self.licensing._entitlements_for_licence(licence), ) def _activation_summary(self, activation) -> JoomlaActivationSummary: suffix = activation.installation_id[-4:].upper() return JoomlaActivationSummary( activationID=activation.id, installationReference=f"Mac installation ****{suffix}", status=activation.status, activatedAt=as_utc(activation.activated_at), lastValidatedAt=as_utc(activation.last_validated_at) if activation.last_validated_at else None, deactivatedAt=as_utc(activation.deactivated_at) if activation.deactivated_at else None, )