package com.example.service.base; import com.example.dto.dialogflow.base.DetectIntentRequestDTO; import com.example.dto.dialogflow.base.DetectIntentResponseDTO; import com.example.exception.DialogflowClientException; import com.example.mapper.conversation.DialogflowRequestMapper; import com.example.mapper.conversation.DialogflowResponseMapper; import com.google.api.gax.rpc.ApiException; import com.google.cloud.dialogflow.cx.v3.DetectIntentRequest; import com.google.cloud.dialogflow.cx.v3.QueryParameters; import com.google.cloud.dialogflow.cx.v3.SessionName; import com.google.cloud.dialogflow.cx.v3.SessionsClient; import com.google.cloud.dialogflow.cx.v3.SessionsSettings; import java.io.IOException; import java.util.Objects; import javax.annotation.PreDestroy; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.stereotype.Service; import reactor.core.publisher.Mono; import reactor.util.retry.Retry; /** * Service for interacting with the Dialogflow CX API to detect user DetectIntent. * It encapsulates the low-level API calls, handling request mapping from DTOs, * managing the `SessionsClient`, and translating API responses into DTOs, * all within a reactive programming context. */ @Service @Qualifier("dialogflowClientService") public class DialogflowClientService implements IntentDetectionService { private static final Logger logger = LoggerFactory.getLogger( DialogflowClientService.class ); private final String dialogflowCxProjectId; private final String dialogflowCxLocation; private final String dialogflowCxAgentId; private final DialogflowRequestMapper dialogflowRequestMapper; private final DialogflowResponseMapper dialogflowResponseMapper; private SessionsClient sessionsClient; public DialogflowClientService( @org.springframework.beans.factory.annotation.Value( "${dialogflow.cx.project-id}" ) String dialogflowCxProjectId, @org.springframework.beans.factory.annotation.Value( "${dialogflow.cx.location}" ) String dialogflowCxLocation, @org.springframework.beans.factory.annotation.Value( "${dialogflow.cx.agent-id}" ) String dialogflowCxAgentId, DialogflowRequestMapper dialogflowRequestMapper, DialogflowResponseMapper dialogflowResponseMapper ) throws IOException { this.dialogflowCxProjectId = dialogflowCxProjectId; this.dialogflowCxLocation = dialogflowCxLocation; this.dialogflowCxAgentId = dialogflowCxAgentId; this.dialogflowRequestMapper = dialogflowRequestMapper; this.dialogflowResponseMapper = dialogflowResponseMapper; try { String regionalEndpoint = String.format( "%s-dialogflow.googleapis.com:443", dialogflowCxLocation ); SessionsSettings sessionsSettings = SessionsSettings.newBuilder() .setEndpoint(regionalEndpoint) .build(); this.sessionsClient = SessionsClient.create(sessionsSettings); logger.info( "Dialogflow CX SessionsClient initialized successfully for endpoint: {}", regionalEndpoint ); logger.info( "Dialogflow CX SessionsClient initialized successfully for agent - Test Agent version: {}", dialogflowCxAgentId ); } catch (IOException e) { logger.error( "Failed to create Dialogflow CX SessionsClient: {}", e.getMessage(), e ); throw e; } } @PreDestroy public void closeSessionsClient() { if (sessionsClient != null) { sessionsClient.close(); logger.info("Dialogflow CX SessionsClient closed."); } } @Override public Mono detectIntent( String sessionId, DetectIntentRequestDTO request ) { Objects.requireNonNull( sessionId, "Dialogflow session ID cannot be null." ); Objects.requireNonNull( request, "Dialogflow request DTO cannot be null." ); logger.info("Initiating detectIntent for session: {}", sessionId); DetectIntentRequest.Builder detectIntentRequestBuilder; try { detectIntentRequestBuilder = dialogflowRequestMapper.mapToDetectIntentRequestBuilder( request ); logger.debug( "Obtained partial DetectIntentRequest.Builder from mapper for session: {}", sessionId ); } catch (IllegalArgumentException e) { logger.error( " Failed to map DTO to partial Protobuf request for session {}: {}", sessionId, e.getMessage() ); return Mono.error( new IllegalArgumentException( "Invalid Dialogflow request input: " + e.getMessage() ) ); } SessionName sessionName = SessionName.newBuilder() .setProject(dialogflowCxProjectId) .setLocation(dialogflowCxLocation) .setAgent(dialogflowCxAgentId) .setSession(sessionId) .build(); detectIntentRequestBuilder.setSession(sessionName.toString()); logger.debug( "Set session path {} on the request builder for session: {}", sessionName.toString(), sessionId ); QueryParameters.Builder queryParamsBuilder; if (detectIntentRequestBuilder.hasQueryParams()) { queryParamsBuilder = detectIntentRequestBuilder .getQueryParams() .toBuilder(); } else { queryParamsBuilder = QueryParameters.newBuilder(); } detectIntentRequestBuilder.setQueryParams(queryParamsBuilder.build()); // Build the final DetectIntentRequest Protobuf object DetectIntentRequest detectIntentRequest = detectIntentRequestBuilder.build(); return Mono.fromCallable(() -> { logger.debug( "Calling Dialogflow CX detectIntent for session: {}", sessionId ); return sessionsClient.detectIntent(detectIntentRequest); }) .retryWhen( reactor.util.retry.Retry.backoff( 3, java.time.Duration.ofSeconds(1) ) .filter(throwable -> { if (throwable instanceof ApiException apiException) { com.google.api.gax.rpc.StatusCode.Code code = apiException.getStatusCode().getCode(); boolean isRetryable = code == com.google.api.gax.rpc.StatusCode.Code.INTERNAL || code == com.google.api.gax.rpc.StatusCode.Code.UNAVAILABLE; if (isRetryable) { logger.warn( "Retrying Dialogflow CX call for session {} due to transient error: {}", sessionId, code ); } return isRetryable; } return false; }) .doBeforeRetry(retrySignal -> logger.debug( "Retry attempt #{} for session {}", retrySignal.totalRetries() + 1, sessionId ) ) .onRetryExhaustedThrow((retrySpec, retrySignal) -> { logger.error( "Dialogflow CX retries exhausted for session {}", sessionId ); return retrySignal.failure(); }) ) .onErrorMap(ApiException.class, e -> { String statusCode = e.getStatusCode().getCode().name(); String message = e.getMessage(); String detailedLog = message; if ( e.getCause() instanceof io.grpc.StatusRuntimeException grpcEx ) { detailedLog = String.format( "Status: %s, Message: %s, Trailers: %s", grpcEx.getStatus().getCode(), grpcEx.getStatus().getDescription(), grpcEx.getTrailers() ); } logger.error( "Dialogflow CX API error for session {}: details={}", sessionId, detailedLog, e ); return new DialogflowClientException( "Dialogflow CX API error: " + statusCode + " - " + message, e ); }) .map(dfResponse -> this.dialogflowResponseMapper.mapFromDialogflowResponse( dfResponse, sessionId ) ); } }