Olympus/backend/hermes/app/services/repair_public_link_service.py

import hashlib
import hmac
import secrets
from datetime import UTC, datetime

from fastapi import HTTPException, status
from sqlalchemy.orm import Session
from starlette.requests import Request

from app.core.config import settings
from app.models.repair import Repair, RepairPublicAccessToken
from app.models.user import User
from app.repositories.repair_repository import RepairRepository
from app.schemas.repair import (
    RepairPublicLinkCreateResponse,
    RepairPublicLinkResponse,
    RepairPublicStatusHistoryItem,
    RepairPublicStatusResponse,
)
from app.services.audit_service import write_audit_log
from app.services.repair_status_labels import STATUS_LABELS
from app.services.system_settings_service import SystemSettingsService


def normalize_token(token: str) -> str:
    return token.strip()


def repair_label(repair: Repair) -> str:
    return f"{repair.repair_number} · {repair.customer_name}"


def public_status_path(db: Session, token: str) -> str:
    normalized_token = normalize_token(token)
    path = f"/status/{normalized_token}"
    base_url = SystemSettingsService.get_public_repair_status_base_url(db)
    if not base_url:
        return path
    return f"{base_url}/{normalized_token}"


class RepairPublicLinkService:
    @staticmethod
    def hash_token(token: str) -> str:
        normalized_token = normalize_token(token)
        return hmac.new(settings.secret_key.encode("utf-8"), normalized_token.encode("utf-8"), hashlib.sha256).hexdigest()

    @staticmethod
    def create_token() -> str:
        return secrets.token_urlsafe(48)

    @staticmethod
    def response_from_link(public_link: RepairPublicAccessToken | None) -> RepairPublicLinkResponse:
        if public_link is None:
            return RepairPublicLinkResponse(is_active=False)
        return RepairPublicLinkResponse(
            is_active=public_link.is_active,
            token_hint=public_link.token_hint,
            expires_at=public_link.expires_at,
            last_used_at=public_link.last_used_at,
            created_at=public_link.created_at,
            revoked_at=public_link.revoked_at,
        )

    @staticmethod
    def get(db: Session, repair: Repair) -> RepairPublicLinkResponse:
        return RepairPublicLinkService.response_from_link(RepairRepository.get_active_public_link(db, repair.id))

    @staticmethod
    def create(db: Session, repair: Repair, *, actor: User, request: Request) -> RepairPublicLinkCreateResponse:
        return RepairPublicLinkService.create_with_audit(
            db,
            repair,
            actor=actor,
            request=request,
            audit_action="repairs.public_link.create",
        )

    @staticmethod
    def create_with_audit(
        db: Session,
        repair: Repair,
        *,
        actor: User | None,
        request: Request | None,
        audit_action: str,
    ) -> RepairPublicLinkCreateResponse:
        token = RepairPublicLinkService.create_token()
        public_link = RepairRepository.create_public_link(
            db,
            repair_id=repair.id,
            token_hash=RepairPublicLinkService.hash_token(token),
            token_hint=token[-6:],
        )
        write_audit_log(
            db,
            action=audit_action,
            entity_type="repairs",
            entity_id=repair.id,
            entity_label=repair_label(repair),
            actor=actor,
            request=request,
            metadata={"token_hint": public_link.token_hint},
        )
        return RepairPublicLinkCreateResponse(
            is_active=True,
            token=token,
            token_hint=public_link.token_hint,
            expires_at=public_link.expires_at,
            last_used_at=public_link.last_used_at,
            created_at=public_link.created_at,
            revoked_at=public_link.revoked_at,
            public_status_path=public_status_path(db, token),
        )

    @staticmethod
    def revoke(db: Session, repair: Repair, *, actor: User, request: Request) -> RepairPublicLinkResponse:
        active_link = RepairRepository.get_active_public_link(db, repair.id)
        RepairRepository.revoke_public_links(db, repair_id=repair.id)
        write_audit_log(
            db,
            action="repairs.public_link.revoke",
            entity_type="repairs",
            entity_id=repair.id,
            entity_label=repair_label(repair),
            actor=actor,
            request=request,
            metadata={"token_hint": active_link.token_hint if active_link else None},
        )
        return RepairPublicLinkResponse(is_active=False, token_hint=active_link.token_hint if active_link else None, revoked_at=datetime.now(UTC))

    @staticmethod
    def public_status(db: Session, token: str) -> RepairPublicStatusResponse:
        normalized_token = normalize_token(token)
        if not normalized_token:
            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Reparaturstatus nicht gefunden")

        public_link = RepairRepository.get_public_link_by_hash(db, RepairPublicLinkService.hash_token(normalized_token))
        if public_link is None or public_link.repair is None:
            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Reparaturstatus nicht gefunden")

        repair = public_link.repair
        history = RepairRepository.get_history_public(db, repair.id)
        response = RepairPublicStatusResponse(
            repair_number=repair.repair_number,
            public_status_label=STATUS_LABELS.get(repair.status, repair.status),
            device_manufacturer=repair.device_manufacturer,
            device_model=repair.device_model,
            status_history_public=[
                RepairPublicStatusHistoryItem(
                    status=item.new_status,
                    status_label=STATUS_LABELS.get(item.new_status, item.new_status),
                    created_at=item.created_at,
                )
                for item in history
            ],
            updated_at=repair.updated_at,
        )
        RepairRepository.mark_public_link_used(db, public_link)
        return response