dify/api/services/message_extra_content_service.py

"""Restore message approvals through their version-specific storage owners."""

import logging
from collections import defaultdict
from collections.abc import Sequence

from sqlalchemy.orm import Session, sessionmaker

from core.entities.execution_extra_content import (
    ExecutionExtraContentDomainModel,
    HumanInputContent,
    HumanInputFormDefinition,
    HumanInputFormSubmissionData,
)
from core.workflow.nodes.human_input_v2.runtime import PreparedForm
from factories.human_input_v2_form_presentation import build_form_presentation, render_filled_content
from libs.datetime_utils import to_utc_timestamp
from repositories.execution_extra_content_repository import ExecutionExtraContentRepository, HumanInputFormReference
from repositories.sqlalchemy_api_workflow_node_execution_repository import (
    DifyAPISQLAlchemyWorkflowNodeExecutionRepository,
)
from services.human_input_v2.form_service import load_prepared_forms

logger = logging.getLogger(__name__)


def load_message_extra_contents(
    *,
    repository: ExecutionExtraContentRepository,
    message_ids: Sequence[str],
    session_factory: sessionmaker[Session],
) -> list[list[ExecutionExtraContentDomainModel]]:
    """The caller admits the messages; preserve their content and form ordering.

    V1 content is already projected by the legacy repository. V2 references
    recover only their original initiator grants, without creating deliveries.
    """
    contents = repository.get_by_message_ids(message_ids)
    references_by_run: dict[str, list[HumanInputFormReference]] = defaultdict(list)
    for message_contents in contents:
        for content in message_contents:
            if isinstance(content, HumanInputFormReference):
                references_by_run[content.workflow_run_id].append(content)

    restored: dict[HumanInputFormReference, HumanInputContent] = {}
    nodes = DifyAPISQLAlchemyWorkflowNodeExecutionRepository(session_maker=session_factory)
    for run_id, references in references_by_run.items():
        prepared_forms = load_prepared_forms(
            workflow_run_id=run_id,
            form_ids=[reference.form_id for reference in references],
            session_factory=session_factory,
        )
        if not prepared_forms:
            logger.warning("No approval forms found for message history: workflow_run_id=%s", run_id)
            continue
        owner = next(iter(prepared_forms.values())).form
        executions = nodes.get_executions_by_workflow_run(str(owner.tenant_id), owner.app_id, run_id)
        node_ids = {execution.id: execution.node_id for execution in executions}
        for reference in references:
            prepared = prepared_forms.get(reference.form_id)
            execution_id = prepared.form.node_execution_id if prepared is not None else None
            node_id = node_ids.get(execution_id) if execution_id is not None else None
            if prepared is None or node_id is None:
                logger.warning(
                    "Approval form or execution missing from message history: workflow_run_id=%s form_id=%s",
                    run_id,
                    reference.form_id,
                )
                continue
            restored[reference] = _restore_v2_content(prepared, workflow_run_id=run_id, node_id=node_id)

    resolved: list[list[ExecutionExtraContentDomainModel]] = []
    for message_contents in contents:
        message_resolved: list[ExecutionExtraContentDomainModel] = []
        for content in message_contents:
            if isinstance(content, HumanInputFormReference):
                if (form_content := restored.get(content)) is not None:
                    message_resolved.append(form_content)
            else:
                message_resolved.append(content)
        resolved.append(message_resolved)
    return resolved


def _restore_v2_content(prepared: PreparedForm, *, workflow_run_id: str, node_id: str) -> HumanInputContent:
    form = prepared.form
    presentation = build_form_presentation(form.resolved_form)
    definition = HumanInputFormDefinition(
        form_id=form.id,
        form_version="2",
        node_id=node_id,
        node_title=form.resolved_form.title,
        form_content=presentation.form_content,
        inputs=presentation.inputs,
        actions=presentation.user_actions,
        resolved_default_values=presentation.resolved_default_values,
        expiration_time=to_utc_timestamp(form.expiration_time),
        display_in_ui=prepared.form_token is not None,
        form_token=prepared.form_token.get_secret_value() if prepared.form_token is not None else None,
    )
    submission = form.submission
    submitted = None
    if submission is not None:
        action_text = next(
            (action.title for action in form.resolved_form.actions if action.id == submission.selected_action_id),
            submission.selected_action_id,
        )
        submitted = HumanInputFormSubmissionData(
            node_id=node_id,
            node_title=form.resolved_form.title,
            rendered_content=render_filled_content(form.resolved_form, submission.normalized_submission_data),
            action_id=submission.selected_action_id,
            action_text=action_text,
            submitted_data=submission.raw_submission_inputs,
        )
    return HumanInputContent(
        workflow_run_id=workflow_run_id,
        submitted=submission is not None,
        form_definition=definition,
        form_submission_data=submitted,
    )