Nexus Oracle Fusion Purchase Order Worker (Service Layer + Alpura-SDK)

9.1 Responsabilidad

Consumir mensajes de Pub/Sub (nexus.purchase-order.requested / retry) y crear la OC en Oracle Fusion Procurement. La integración se encapsula mediante interfaces Java (PurchaseOrderCreationService) en la capa Facade e implementaciones en la capa Service usando Alpura-SDK (HTTP clients, OAuth client credentials, Resilience4j, OpenTelemetry tracing, serialización, secrets). No se usa la terminología Port/Adapter de Hexagonal.

9.2 Ficha técnica del componente

Aspecto	Detalle
Responsabilidad	Consumir, idempotencia, transformación canónico→Fusion, invocación REST, actualización estado, retry/DLQ, auditoría.
Entradas	Envelope Pub/Sub (JSON).
Salidas	Estado en PostgreSQL + audit_event + datos Oracle (OC number).
Dependencias	PostgreSQL, Oracle Fusion REST API, Secret Manager (credenciales Fusion), IdP token endpoint.
Datos administrados	purchase_order, purchase_order_line, transaction_status_history, integration_error, processed_event.
APIs	Sin APIs HTTP; consumer suscripción (Spring Cloud GCP PubSubMessageController en capa WEB equivalente).
Eventos	audit_event varios tipos.
Estados	Transiciona QUEUED → PROCESSING → CREATED/FAILED/RETRY_PENDING/DLQ.
Errores	Taxonomía: TEMPORARY_ERROR, INTEGRATION_ERROR, BUSINESS_ERROR, TECHNICAL, NON_RETRYABLE.
Patrones	Idempotent Consumer, Service Facade + Alpura-SDK (Resilience4j CB/Bulkhead/Retry/Timelimiter).
Seguridad	Credenciales por partner; scopes; rotation secrets TBD.
Observabilidad	Latencia Fusion, reintentos, DLQ, hist por partner.

9.3 Diseño por capas (Arquitectura Base Backend Alpura)

Capa	Clases / Interfaces	Métodos (nomenclatura Alpura)
WEB (Listener Pub/Sub)	PubSubMessageController (Spring Cloud GCP)	onRequestedMessage(message), onRetryMessage(message), onDlqMessage(message)
Facade (Service Interfaces)	PurchaseOrderCreationService (interface)	createPurchaseOrder(request) (1 parámetro Value Object CreatePurchaseOrderServiceRequest)
Service (implementaciones)	OracleFusionPurchaseOrderService, FusionTokenResolver, FusionMappingService	Públicos: createPurchaseOrder() Privados (prefijo _): _getFusionAccessToken(), _resolveMappingsForPartner(), _mapCanonicalToFusionRequest(), _invokeFusionCreatePo(), _classifyFusionResponseError(), _appendAudit(), _registerProcessedEvent()
Persistence	PurchaseOrderRepository, PurchaseOrderLineRepository, ProcessedEventRepository, TransactionStatusRepository, IntegrationErrorRepository, AuditRepository	Públicos: findByRequestId(), saveRequestAndOutboxTx(), findByIdempotencyKey(), saveIntegrationError(), markProcessed() Privados (en impl custom): _mapToEntity(), _buildCriteriaQuery()
Commons	commons.valueobjects (FusionClientConfig, ScopeContext, Correlation, OraclePoNumber, FusionPoId) commons.exceptions (FusionAuthenticationException, FusionBusinessException, FusionTemporaryException, FusionIntegrationException) commons.constants (FusionHeaderKeys, FusionStatusCodes, ResilienceDefaults)

9.4 Flujo interno (pseudocódigo con nomenclatura Alpura)

// WEB / Listener equivalente (Spring Cloud GCP)
@RestController
public class PubSubMessageController {

    private final PurchaseOrderCreationService creationService;   // Facade interface

    @PubSubListener(subscription = "nexus.purchase-order.requested.sub")
    public void onRequestedMessage(PubsubMessage message) {
        Envelope envelope = _parseEnvelope(message);
        // 1) Idempotent consumer
        if (processedEventRepo.exists(envelope.getEventId())) {
            acknowledge(message); return;
        }
        // 2) carga request + canonical
        PurchaseOrderRequest req = purchaseOrderRepo.findByRequestId(envelope.getRequestId())
                .orElseThrow(() -> new NexusNotFoundException("requestId", envelope.getRequestId()));
        // 3) valida estado permisible
        if (!ALLOWED_STATUSES.contains(req.getStatus())) {
            processedEventRepo.markProcessed(envelope.getEventId(), DUPLICATE_SKIPPED);
            acknowledge(message); return;
        }
        txHelper.executeInTx(() -> {
            req.transitionStatus(PROCESSING, "worker starting fusion");
            statusRepo.appendTransition(req.getRequestId(), PROCESSING, "fusion-worker", null);
        });

        // 4) invoca capa Service mediante interface Facade (1 param Value Object)
        CreatePurchaseOrderServiceRequest serviceRequest = _toServiceRequest(req, envelope);
        try {
            CreatePurchaseOrderServiceResponse result = creationService.createPurchaseOrder(serviceRequest);
            _handleCreated(req, envelope, result);
        } catch (NexusTemporaryException e) {
            _handleTemporary(req, envelope, e);
            return;
        } catch (NexusNonRetryableException e) {
            _handleNonRetryable(req, envelope, e);
            return;
        }
        acknowledge(message);
    }

    private Envelope _parseEnvelope(PubsubMessage m) { /* ... */ }
    private CreatePurchaseOrderServiceRequest _toServiceRequest(PurchaseOrderRequest r, Envelope env) { /* agrupa */ }
    private void _handleCreated(PurchaseOrderRequest r, Envelope e, CreatePurchaseOrderServiceResponse res) { /* ... */ }
    private void _handleTemporary(PurchaseOrderRequest r, Envelope e, NexusTemporaryException ex) { /* pub retry topic */ }
    private void _handleNonRetryable(PurchaseOrderRequest r, Envelope e, NexusNonRetryableException ex) { /* pub DLQ */ }
}

9.5 Integración con Oracle Fusion (conceptual)

TBD — pendiente de confirmar con documentación oficial de Oracle Fusion: endpoint específico, versión API, nombre de recurso, IDs de BU/LE, esquema de payload, estructura success response y códigos error.

9.5.1 Puntos acordados TO-BE (antes de confirmación con Fusion)

Aspecto	Acuerdo provisional
Estilo API	REST (documentado en docs Oracle Fusion Procurement: Purchase Orders / requisitions / import batch). TBD.
Autenticación	OAuth 2.0 client credentials contra token endpoint de Fusion. TBD. Token obtenido por FusionTokenResolver usando Alpura-SDK + Secret Manager.
Timeout	Resilience4j TimeLimiter: TBD 30s connect / TBD 90s read. Configurado por medio de Alpura-SDK.webClientBuilder().
Retries	Exponential backoff TBD (initial=2s, factor=2, max=30s, attempts=3 en _invokeFusionCreatePo; luego delegar a Pub/Sub retry policy).
Circuit Breaker	Resilience4j CB por partner + endpoint. Sliding window count. TBD umbrales.
Bulkhead	Semáforo por partner (TBD maxConcurrentCalls) para proteger Fusion.
Errores funcionales	Cualquier respuesta 4xx con error code de negocio → NON_RETRYABLE → FAILED/DLQ (según configuración).
Errores técnicos	5xx / timeout / socket exception → TEMPORARY_ERROR → retry / DLQ.
Mapping	Realizado en FusionMappingService; mapeos específicos por partnerId via partner_configuration JSONB.
Idempotencia Fusion	Pasar externalReference / requestId como campo de request id si Fusion admite (ej: Attribute1, campo flexfield, o partnerReference). TBD nombre campo exacto.

9.5.2 Interface Service Facade y implementación (Alpura-SDK)

// Facade: Service Interface (capa Facade)
public interface PurchaseOrderCreationService {
    // 1 param DT (agrupa canonical + context) -> 0-3 params
    CreatePurchaseOrderServiceResponse createPurchaseOrder(CreatePurchaseOrderServiceRequest request)
            throws NexusTemporaryException, NexusNonRetryableException;
}

// Value Object (Commons · Value Objects) - reduce params
public record CreatePurchaseOrderServiceRequest(
        CanonicalPurchaseOrder canonical,
        ScopeContext scope,
        Correlation correlation,
        FusionClientConfig fusionConfig
) {}

// Service impl - capa Service (implementa facade; usa Alpura-SDK)
@Service
public class OracleFusionPurchaseOrderService implements PurchaseOrderCreationService {

    private final AlpuraSdk alpuraSdk;   // Libreria Alpura morada
    private final FusionMappingService mappingService;
    private final AuditWriter auditWriter;

    @Override
    public CreatePurchaseOrderServiceResponse createPurchaseOrder(CreatePurchaseOrderServiceRequest req) {
        FusionClientConfig cfg = req.fusionConfig() != null
                ? req.fusionConfig()
                : _resolveFusionConfig(req.scope());
        String accessToken = _getFusionAccessToken(cfg);
        Map<String,Object> fusionPayload = _mapCanonicalToFusionRequest(req.canonical(), cfg);
        FusionHttpResponse response = _invokeFusionCreatePo(cfg, accessToken, fusionPayload);
        if (!response.isSuccessful()) {
            var classified = _classifyFusionResponseError(response, req);
            _appendAudit(req, classified);
            throw classified.exception();
        }
        FusionPoId fusionPoId = response.extractPoId();
        _registerProcessedEvent(req.correlation().eventId(), fusionPoId);
        _appendAudit(req, FusionOutcome.builder().poId(fusionPoId).build());
        return new CreatePurchaseOrderServiceResponse(req.correlation().requestId(), fusionPoId);
    }

    private FusionClientConfig _resolveFusionConfig(ScopeContext scope) { /* ... */ }
    private String _getFusionAccessToken(FusionClientConfig cfg) { /* Alpura-SDK + Secret Manager */ }
    private Map<String,Object> _mapCanonicalToFusionRequest(CanonicalPurchaseOrder canonical, FusionClientConfig cfg) {
        return mappingService.toFusionPayload(canonical, cfg);
    }
    private FusionHttpResponse _invokeFusionCreatePo(
            FusionClientConfig cfg,
            String token,
            Map<String,Object> payload
    ) {
        var webClient = alpuraSdk.httpClientBuilder()
                .withBaseUrl(cfg.baseUrl())
                .withAuthentication(OAuth2Bearer.of(token))
                .withCircuitBreaker(CircuitBreakerConfig.forName("fusion-" + cfg.partnerId()))
                .withBulkhead(BulkheadConfig.semaphore("fusion-" + cfg.partnerId()))
                .withTimeLimiter(Duration.ofSeconds(90))
                .withTracing(propagateTraceparent())
                .build();
        return webClient.post(cfg.createPoPath())
                .body(payload)
                .retrieve()
                .toEntity(FusionHttpResponse.class);
    }
    private ClassifiedFusionError _classifyFusionResponseError(FusionHttpResponse resp, CreatePurchaseOrderServiceRequest req) { /* ... */ }
    private void _appendAudit(CreatePurchaseOrderServiceRequest req, FusionOutcome outcome) { /* ... */ }
    private void _registerProcessedEvent(UUID eventId, FusionPoId poId) { /* ... */ }
}

9.6 Futuras implementaciones Service (extensibilidad)

En lugar de “adapters del puerto”, se agregan nuevas implementaciones de la interface PurchaseOrderCreationService y se selecciona la instancia correcta por medio de @Qualifier por partner / sourceSystem:

Selección: PurchaseOrderServiceFactory.getForPartner(partnerId) devuelve la implementación correcta. Todas usan Alpura-SDK para su cliente HTTP, auth y resiliencia.