Aegis/backend/app/services/tempo_service.py

"""Tempo time-tracking integration service.

Authentication model
--------------------
Each user authenticates to Tempo with their own personal Tempo API token,
stored in ``user.tempo_api_token``.  This is different from the Jira API token.
Obtain a Tempo token at: Jira → Apps → Tempo → Settings → API Integration.

The global ``settings.TEMPO_ENABLED`` flag acts as a kill-switch.  When False,
all Tempo calls are silently skipped regardless of whether users have tokens.
"""

import logging
from typing import Optional

from sqlalchemy.orm import Session

from app.config import settings
from app.domain.exceptions import InvalidOperationError
from app.models.jira_link import JiraLink, JiraLinkEntityType

logger = logging.getLogger(__name__)


def has_tempo_configured(user) -> bool:
    """Return True if *user* has a personal Tempo API token stored."""
    return bool(getattr(user, "tempo_api_token", None))


def get_user_tempo_client(user):
    """Return a Tempo API v4 client authenticated as *user*.

    Raises ``InvalidOperationError`` when the user has no token or the
    client library is not installed.
    """
    token = getattr(user, "tempo_api_token", None)
    if not token:
        raise InvalidOperationError(
            "No Tempo API token configured. "
            "Add it in Settings → Profile → Tempo Integration."
        )
    try:
        from tempoapiclient import client_v4 as tempo_client
        return tempo_client.Tempo(auth_token=token)
    except ImportError:
        raise InvalidOperationError(
            "tempo-api-python-client is not installed. "
            "Run: pip install tempo-api-python-client"
        )


def log_worklog(
    user,
    jira_issue_id: int,
    author_account_id: str,
    date: str,
    time_spent_seconds: int,
    description: str,
    work_type: str | None = None,
) -> dict:
    """Create a worklog entry in Tempo using *user*'s personal token."""
    tempo = get_user_tempo_client(user)
    kwargs: dict = {
        "accountId": author_account_id,
        "issueId": jira_issue_id,
        "dateFrom": date,
        "timeSpentSeconds": time_spent_seconds,
        "description": description,
    }
    wt = work_type or settings.TEMPO_DEFAULT_WORK_TYPE
    if wt:
        kwargs["workType"] = wt
    return tempo.create_worklog(**kwargs)


def auto_log_test_worklog(
    db: Session,
    test,
    user,
    activity_type: str,
) -> Optional[dict]:
    """If the test has a Jira link and *user* has a Tempo token, log time.

    Returns the Tempo worklog response, or None if skipped.
    Completely non-fatal — errors are logged and swallowed.
    """
    # Global kill-switch
    if not settings.TEMPO_ENABLED:
        return None

    # Per-user token required
    if not has_tempo_configured(user):
        logger.debug(
            "User %s has no Tempo token; skipping worklog for test %s",
            getattr(user, "username", user), test.id,
        )
        return None

    # Need a Jira link with a numeric issue ID
    link = (
        db.query(JiraLink)
        .filter(
            JiraLink.entity_id == test.id,
            JiraLink.entity_type == JiraLinkEntityType.test,
        )
        .first()
    )

    if not link or not link.jira_issue_id:
        logger.debug("No Jira link for test %s, skipping Tempo worklog", test.id)
        return None

    jira_account_id = getattr(user, "jira_account_id", "") or ""
    if not jira_account_id:
        logger.debug(
            "User %s has no jira_account_id; skipping Tempo worklog",
            getattr(user, "username", user),
        )
        return None

    duration = _calculate_duration(test, activity_type)
    if duration <= 0:
        return None

    try:
        result = log_worklog(
            user=user,
            jira_issue_id=int(link.jira_issue_id),
            author_account_id=jira_account_id,
            date=(getattr(test, "updated_at", None) or test.created_at).strftime("%Y-%m-%d"),
            time_spent_seconds=duration,
            description=f"[Aegis] {activity_type}: {test.name}",
        )
        logger.info(
            "Tempo worklog created for test %s by user %s, %ds",
            test.id, getattr(user, "username", user), duration,
        )
        return result
    except Exception as e:
        logger.warning(
            "Tempo worklog failed for test %s (user %s): %s",
            test.id, getattr(user, "username", user), e, exc_info=True,
        )
        return None


def _calculate_duration(test, activity_type: str) -> int:
    """Calculate real duration in seconds from the phase timing fields.

    Uses the actual start/end timestamps recorded by the workflow buttons,
    so the data cannot be falsified.
    """
    from datetime import datetime

    now = datetime.utcnow()

    if activity_type == "red_team_execution" and test.red_started_at:
        delta = now - test.red_started_at
        return max(int(delta.total_seconds()), 1)

    if activity_type == "blue_team_evaluation" and test.blue_started_at:
        delta = now - test.blue_started_at
        return max(int(delta.total_seconds()), 1)

    # Fallback for legacy activity types
    if activity_type == "execution" and test.execution_date and test.created_at:
        delta = test.execution_date - test.created_at
        return max(int(delta.total_seconds()), 0)

    return 0