Integračná príručka QHUB — Peppol dokumenty cez RabbitMQ.

POST {QHUB_REGISTRATION_URL}/api/v1/SyncEntityCatalogItem
Content-Type: application/json
Accept: application/json
Authorization: {API kľúč}

{
  "System": "MYERP",
  "SyncEntityDatas": [
    {
      "EntityCode": "PeppolDocument",
      "EntityType": 30,
      "DefaultAllowed": true,
      "Version": "1.0.0",
      "SyncEntityDefinitionData": {
        "SyncPropertiesData": [
          { "Name": "SenderParticipantId",   "OriginalName": "SenderParticipantId",   "ProperyType": "String",   "Required": true,  "IsBusinessKey": false },
          { "Name": "ReceiverParticipantId", "OriginalName": "ReceiverParticipantId", "ProperyType": "String",   "Required": true,  "IsBusinessKey": false },
          { "Name": "DocumentTypeId",        "OriginalName": "DocumentTypeId",        "ProperyType": "String",   "Required": true,  "IsBusinessKey": false },
          { "Name": "ProcessId",             "OriginalName": "ProcessId",             "ProperyType": "String",   "Required": true,  "IsBusinessKey": false },
          { "Name": "UblDocument",           "OriginalName": "UblDocument",           "ProperyType": "String",   "Required": true,  "IsBusinessKey": false },
          { "Name": "As4MessageId",          "OriginalName": "As4MessageId",          "ProperyType": "String",   "Required": false, "IsBusinessKey": true  },
          { "Name": "ReceivedAt",            "OriginalName": "ReceivedAt",            "ProperyType": "DateTime", "Required": false, "IsBusinessKey": false }
        ],
        "SyncForeignKeysData": []
      }
    },
    {
      "EntityCode": "PeppolDocumentEvent",
      "EntityType": 32,
      "DefaultAllowed": true,
      "Version": "1.0.0",
      "SyncEntityDefinitionData": {
        "SyncPropertiesData": [
          { "Name": "EventId",                            "OriginalName": "eventId",                            "ProperyType": "String",   "Required": true,  "IsBusinessKey": true  },
          { "Name": "PeppolDocumentMessageCorrelationId", "OriginalName": "peppolDocumentMessageCorrelationId", "ProperyType": "String",   "Required": true,  "IsBusinessKey": false },
          { "Name": "EventType",                          "OriginalName": "EventType",                          "ProperyType": "String",   "Required": true,  "IsBusinessKey": false },
          { "Name": "Message",                            "OriginalName": "Message",                            "ProperyType": "String",   "Required": false, "IsBusinessKey": false },
          { "Name": "Error",                              "OriginalName": "Error",                              "ProperyType": "String",   "Required": false, "IsBusinessKey": false },
          { "Name": "ValidationErrors",                   "OriginalName": "ValidationErrors",                   "ProperyType": "String",   "Required": false, "IsBusinessKey": false },
          { "Name": "ValidationWarnings",                 "OriginalName": "ValidationWarnings",                 "ProperyType": "String",   "Required": false, "IsBusinessKey": false },
          { "Name": "Timestamp",                          "OriginalName": "Timestamp",                          "ProperyType": "DateTime", "Required": true,  "IsBusinessKey": false }
        ],
        "SyncForeignKeysData": [
          { "Name": "PeppolDocumentKey", "EntityType": 30, "Required": true }
        ]
      }
    }
  ]
}

POST {QHUB_REGISTRATION_URL}/api/v1/Registration/Validate
Content-Type: application/json

{
  "participantSystem": "MYERP",
  "participantTenantId": "c0ffee00-1111-2222-3333-444455556666",
  "registrationKey": "REG-KEY-OD-ASOL",
  "routingKey": "MYERP.c0ffee00-1111-2222-3333-444455556666"
}

{
  "participantSystem": "MYERP",
  "participantTenantId": "c0ffee00-1111-2222-3333-444455556666",
  "rabbitMqUrl": "amqps://messaging.qasida.sk:5671",
  "rabbitMqName": "myerp_c0ffee00",
  "rabbitMqPass": "••••••••••••",
  "errorCode": null,
  "success": true,
  "registeredEntities": [ … ]
}

  VY publikujete dokument                          VY čítate doručenia a eventy
          │                                                   ▲
          ▼                                                   │
┌──────────────────────┐                     ┌────────────────────────────────────┐
│ clouderp-sync-queue  │                     │ queue clouderp-delivery-queue.{RK} │
│ (fanout exchange)    │                     └─────────────────▲──────────────────┘
└──────────┬───────────┘                                       │ binding (fanout, "")
           ▼                                 ┌─────────────────┴──────────────────┐
     ┌───────────┐                           │ fanout exchange                    │
     │   QHUB    │──────────────────────────▶│ clouderp-delivery-queue.{RK}       │
     │ Sync Svc  │  publikuje do             └─────────────────▲──────────────────┘
     └─────▲─────┘  clouderp-delivery-queue                    │ exchange-to-exchange
           │        (direct, routing key = {RK})               │ binding, routing key = {RK}
           │                                 ┌─────────────────┴──────────────────┐
           │                                 │ direct exchange                    │
           │                                 │ clouderp-delivery-queue            │
           │                                 └────────────────────────────────────┘
┌──────────┴───────────────────┐
│ clouderp-sync-response-queue │  ◀── VY publikujete ACK
│ (fanout exchange)            │
└──────────────────────────────┘

1) fanout exchange   clouderp-delivery-queue.{RoutingKey}   durable=true, auto-delete=false
2) queue             clouderp-delivery-queue.{RoutingKey}   durable=true, exclusive=false, auto-delete=false
                     arguments: x-dead-letter-exchange = clouderp.dlq
3) queue binding     queue clouderp-delivery-queue.{RoutingKey}  ←  exchange clouderp-delivery-queue.{RoutingKey}
                     routing key ""
4) exchange binding  destination clouderp-delivery-queue.{RoutingKey} (EXCHANGE)
                     source      clouderp-delivery-queue (direct)
                     routing key {RoutingKey}

Peppol odosielateľ → AS4 → APEX
  → validácia (XSD + Schematron) a uloženie; odosielateľovi ide Peppol MLS
  → len platný dokument: APEX publikuje SynchronizationMessage do clouderp-sync-queue (cca do 5 s)
    → QHUB vytvorí doručenie pre každý zaregistrovaný cieľový systém tenanta
      → clouderp-delivery-queue (direct, routing key = váš RoutingKey)
        → clouderp-delivery-queue.{RoutingKey} (vaša fanout exchange) → vaša queue
          → VÁŠ KONZUMENT → import → ACK do clouderp-sync-response-queue

{
  "messageId": "8f14e45f-ceea-467a-9575-9c4b9b8a1c3d",
  "correlationId": "3b2f1c0a-9d8e-4f7a-b6c5-1a2b3c4d5e6f",
  "messageType": [
    "urn:message:CloudERP.QHubApi.Hub.Abstractions.Messages:SynchronizationMessage"
  ],
  "sentTime": "2026-10-09T08:15:31.1234567Z",
  "headers": {},
  "host": { "machineName": "…", "processName": "…", "massTransitVersion": "…" },
  "message": {
    "transactionId": 4711,
    "eventType": 1,
    "entityType": 30,
    "routingKey": "MYERP.c0ffee00-1111-2222-3333-444455556666",
    "ownerTenantId": "c0ffee00-1111-2222-3333-444455556666",
    "ownerSystem": "APEX",
    "deliverySystem": "MYERP",
    "deliveryTenantId": "c0ffee00-1111-2222-3333-444455556666",
    "syncKey": "abc123@ap.sender.example",
    "schemaVersion": "1.0.0",
    "environment": "…",
    "domainSyncVersion": 1,
    "occurredAt": "2026-10-09T08:15:30.5Z",
    "data": "{\"SenderParticipantId\":\"iso6523-actorid-upis::0245:2024000002\", … ,\"UblDocument\":\"<StandardBusinessDocument …>…\"}",
    "ignoredProperties": null,
    "suffixRoutingKey": null
  }
}

{
  "messageId": "5d0b3c9e-2f7a-5c1b-9e44-0a6f2d1c8b77",
  "correlationId": "3b2f1c0a-9d8e-4f7a-b6c5-1a2b3c4d5e6f",
  "messageType": [
    "urn:message:CloudERP.QHubApi.Hub.Abstractions.Messages:SynchronizationResponseMessage"
  ],
  "sentTime": "2026-10-09T08:15:35Z",
  "message": {
    "transactionId": 4711,
    "result": { "success": true, "errorCode": null, "errorParams": null }
  }
}

VÁŠ SYSTÉM → SynchronizationMessage (entityType 30) → clouderp-sync-queue
  → QHUB → delivery queue APEX-u
    → APEX (spracovanie každé ~3 s): kontroly 1–10 (6.4)
       ├─ zamietnutie  → PeppolDocumentEvent VALIDATION_ERROR / LICENSE_ERROR / AUTHORIZATION_ERROR / PROCESSING_ERROR
       └─ prevzatie    → MESSAGE_QUEUED (+ ACK do QHUB)
            → AS4 odoslanie s opakovaním → MESSAGE_DELIVERED | QUEUE_PROCESSING_FAILED … | QUEUE_DEAD_LETTER
                 → neskôr prípadne MLS_RECEIVED / MLS_RECEIVED_NEGATIVE
APEX → PeppolDocumentEvent (entityType 32) → QHUB → vaša delivery queue (alebo queue prevádzky pri suffixRoutingKey)

{
  "transactionId": 0,
  "eventType": 1,
  "entityType": 30,
  "routingKey": null,
  "ownerTenantId": "c0ffee00-1111-2222-3333-444455556666",
  "ownerSystem": "MYERP",
  "deliverySystem": null,
  "deliveryTenantId": "c0ffee00-1111-2222-3333-444455556666",
  "syncKey": "9a1d2c3e-6b7f-4c8d-9e0a-1b2c3d4e5f60",
  "schemaVersion": "1.0.0",
  "environment": "Production",
  "domainSyncVersion": 1,
  "occurredAt": "2026-10-09T08:15:30Z",
  "data": "{\"SenderParticipantId\":\"0245:2024000001\",\"ReceiverParticipantId\":\"0245:2024000002\",\"DocumentTypeId\":\"urn:oasis:names:specification:ubl:schema:xsd:Invoice-2::Invoice##urn:cen.eu:en16931:2017#compliant#urn:fdc:peppol.eu:2017:poacc:billing:3.0::2.1\",\"ProcessId\":\"urn:fdc:peppol.eu:2017:poacc:billing:01:1.0\",\"UblDocument\":\"<?xml version=\\\"1.0\\\" encoding=\\\"UTF-8\\\"?><Invoice xmlns=…>…</Invoice>\"}",
  "ignoredProperties": null,
  "suffixRoutingKey": "SN-001"
}

úspech:        MESSAGE_QUEUED → QUEUE_PROCESSING_STARTED → MESSAGE_DELIVERED (→ MLS_RECEIVED)
opakovanie:    MESSAGE_QUEUED → QUEUE_PROCESSING_STARTED → QUEUE_PROCESSING_FAILED → … → MESSAGE_DELIVERED
neúspech:      MESSAGE_QUEUED → … → QUEUE_PROCESSING_FAILED (4×) → QUEUE_DEAD_LETTER
zamietnutie:   VALIDATION_ERROR | LICENSE_ERROR | AUTHORIZATION_ERROR | PROCESSING_ERROR   (nič sa nezaradilo)
neskôr:        MESSAGE_DELIVERED → MLS_RECEIVED_NEGATIVE   (prijímajúci AP dokument odmietol)

{
  "transactionId": 4720,
  "eventType": 1,
  "entityType": 32,
  "routingKey": "MYERP.c0ffee00-1111-2222-3333-444455556666",
  "ownerTenantId": "c0ffee00-1111-2222-3333-444455556666",
  "ownerSystem": "APEX",
  "deliverySystem": "MYERP",
  "deliveryTenantId": "c0ffee00-1111-2222-3333-444455556666",
  "syncKey": "2f6d0a1e-4b3c-4d5e-8f70-a1b2c3d4e5f6",
  "schemaVersion": "1.0.0",
  "environment": "…",
  "domainSyncVersion": 1,
  "occurredAt": "2026-10-09T08:15:36Z",
  "data": "{\"EventId\":\"f47ac10b-58cc-4372-a567-0e02b2c3d479\",\"EventType\":\"VALIDATION_ERROR\", … }",
  "ignoredProperties": null,
  "suffixRoutingKey": "SN-001"
}

{
  "EventId": "f47ac10b-58cc-4372-a567-0e02b2c3d479",
  "PeppolDocumentMessageCorrelationId": "3b2f1c0a-9d8e-4f7a-b6c5-1a2b3c4d5e6f",
  "EventType": "VALIDATION_ERROR",
  "Message": "Document send rejected — document validation failed",
  "Error": "Document validation failed: 1 error(s)",
  "ValidationErrors": "[{\"code\":\"PEPPOL-EN16931-R003\",\"message\":\"A buyer reference or purchase order reference MUST be provided.\",\"location\":\"/Invoice\"}]",
  "ValidationWarnings": null,
  "Timestamp": "2026-10-09T10:15:31.123456789+02:00",
  "PeppolDocumentKey": {
    "BusinessKey": null,
    "SyncKey": "9a1d2c3e-6b7f-4c8d-9e0a-1b2c3d4e5f60"
  }
}

{
  "SenderParticipantId": "iso6523-actorid-upis::0245:2024000002",
  "ReceiverParticipantId": "iso6523-actorid-upis::0245:2024000001",
  "DocumentTypeId": "busdox-docid-qns::urn:oasis:names:specification:ubl:schema:xsd:Invoice-2::Invoice##urn:cen.eu:en16931:2017#compliant#urn:fdc:peppol.eu:2017:poacc:billing:3.0::2.1",
  "ProcessId": "urn:fdc:peppol.eu:2017:poacc:billing:01:1.0",
  "UblDocument": "<StandardBusinessDocument xmlns=\"http://www.unece.org/cefact/namespaces/StandardBusinessDocumentHeader\"><StandardBusinessDocumentHeader>…</StandardBusinessDocumentHeader><Invoice xmlns=\"urn:oasis:names:specification:ubl:schema:xsd:Invoice-2\">…</Invoice></StandardBusinessDocument>",
  "As4MessageId": "abc123@ap.sender.example",
  "ReceivedAt": "2026-10-09T08:15:29.871Z"
}

ROUTING_KEY            = SYSTEM_CODE + "." + TENANT_ID
EXCHANGE_DELIVERY      = "clouderp-delivery-queue"          (direct, existuje)
EXCHANGE_SYNC          = "clouderp-sync-queue"              (fanout, existuje)
EXCHANGE_SYNC_RESPONSE = "clouderp-sync-response-queue"     (fanout, existuje)
DELIVERY               = "clouderp-delivery-queue." + ROUTING_KEY   (vaša fanout exchange aj queue)
MT_SYNC     = "urn:message:CloudERP.QHubApi.Hub.Abstractions.Messages:SynchronizationMessage"
MT_RESPONSE = "urn:message:CloudERP.QHubApi.Hub.Abstractions.Messages:SynchronizationResponseMessage"
ENTITY_PEPPOL_DOCUMENT = 30, ENTITY_PEPPOL_DOCUMENT_EVENT = 32

@Configuration
@EnableRabbit
public class QHubConfig {

    static final String EXCHANGE_DELIVERY = "clouderp-delivery-queue";   // direct, existuje

    @Value("${qhub.routing-key}") String routingKey;    // MYERP.{TenantId}
    @Value("${qhub.url}") String url;                   // amqps://messaging.qasida.sk:5671
    @Value("${qhub.username}") String username;
    @Value("${qhub.password}") String password;
    @Value("${qhub.ca-cert-path}") String caCertPath;   // CA brokera (PEM)

    String deliveryName() { return EXCHANGE_DELIVERY + "." + routingKey; }

    @Bean
    CachingConnectionFactory qhubConnectionFactory() throws Exception {
        CachingConnectionFactory cf = new CachingConnectionFactory();
        cf.setUri(url);
        cf.setUsername(username);
        cf.setPassword(password);
        // SSLContext s truststore z CA brokera (pri mTLS aj s klientskym RSA kľúčom)
        cf.getRabbitConnectionFactory().useSslProtocol(sslContextFromPem(caCertPath));
        return cf;
    }

    @Bean
    RabbitAdmin qhubAdmin(CachingConnectionFactory cf) { return new RabbitAdmin(cf); }

    // Deklarujú sa len vlastné objekty. Centrálne exchange ako @Bean NEDEFINUJTE —
    // RabbitAdmin by ich deklaroval a skončil s ACCESS_REFUSED.
    @Bean
    FanoutExchange myDeliveryExchange() { return new FanoutExchange(deliveryName(), true, false); }

    @Bean
    Queue myDeliveryQueue() {
        return QueueBuilder.durable(deliveryName())
                .withArgument("x-dead-letter-exchange", "clouderp.dlq")
                .build();
    }

    @Bean
    Binding queueToMyExchange() {
        return BindingBuilder.bind(myDeliveryQueue()).to(myDeliveryExchange());
    }

    /** Exchange-to-exchange binding: bez neho do queue nikdy nič nepríde. */
    @Bean
    Binding myExchangeToCentral() {
        return new Binding(deliveryName(), Binding.DestinationType.EXCHANGE,
                EXCHANGE_DELIVERY, routingKey, null);
    }

    @Bean
    SimpleRabbitListenerContainerFactory qhubListenerFactory(CachingConnectionFactory cf) {
        SimpleRabbitListenerContainerFactory f = new SimpleRabbitListenerContainerFactory();
        f.setConnectionFactory(cf);
        f.setPrefetchCount(10);
        f.setAcknowledgeMode(AcknowledgeMode.AUTO);
        f.setDefaultRequeueRejected(false);   // výnimka → DLX, nie nekonečný requeue
        // žiadny MessageConverter: listener dostane surové bajty MassTransit obálky
        return f;
    }

    @Bean
    RabbitTemplate qhubTemplate(CachingConnectionFactory cf) { return new RabbitTemplate(cf); }
}

@Component
@RequiredArgsConstructor
public class QHubDeliveryListener {

    private final ObjectMapper mapper;        // FAIL_ON_UNKNOWN_PROPERTIES = false
    private final InboxRepository inbox;      // vaša tabuľka s unikátnymi indexmi

    // Listener len uloží surovú správu; spracovanie robí QHubInboxProcessor.
    @RabbitListener(queues = "#{@myDeliveryQueue.name}", containerFactory = "qhubListenerFactory")
    public void onDelivery(Message amqp) throws IOException {
        byte[] body = amqp.getBody();
        if (body.length == 0) {
            return;
        }
        JsonNode envelope = mapper.readTree(body);
        if (envelope.path("messageType").toString().contains("SynchronizationResponseMessage")
                || !envelope.hasNonNull("message")) {
            return;                           // ACK obálka alebo prázdna správa: neukladať, neodpovedať
        }
        inbox.saveIfNew(                      // pri duplicite (transactionId / correlationId) nič neurobí
                envelope.path("message").path("transactionId").asLong(),
                envelope.path("correlationId").asText(null),
                new String(body, StandardCharsets.UTF_8));
    }
}

@Slf4j
@Component
@RequiredArgsConstructor
public class QHubInboxProcessor {

    private final ObjectMapper mapper;
    private final InboxRepository inbox;
    private final QHubAckSender ack;

    @Scheduled(fixedDelay = 3000)
    public void process() throws IOException {
        for (InboxRow row : inbox.findPending(50)) {
            JsonNode msg = mapper.readTree(row.rawJson()).path("message");
            String entityType = msg.path("entityType").asText();           // číslo aj reťazec
            JsonNode data = mapper.readTree(msg.path("data").asText());     // 2. deserializácia, PascalCase
            try {
                if (entityType.equals("30") || entityType.equals("PeppolDocument")) {
                    importDocument(data);     // UblDocument = StandardBusinessDocument, ID v kanonickom tvare
                } else if (entityType.equals("32") || entityType.equals("PeppolDocumentEvent")) {
                    updateSentDocument(data.path("PeppolDocumentKey").path("SyncKey").asText(),
                            data.path("EventType").asText(), data);
                } else {
                    log.warn("Neznámy entityType {}", entityType);
                }
                ack.send(row.correlationId(), row.transactionId(), true, null, null);
            } catch (Exception e) {
                ack.send(row.correlationId(), row.transactionId(), false, "ImportFailed",
                        List.of(String.valueOf(row.transactionId())));
            }
            inbox.markProcessed(row.id());
        }
    }
}

@Component
@RequiredArgsConstructor
public class QHubAckSender {

    private static final String MT_RESPONSE =
            "urn:message:CloudERP.QHubApi.Hub.Abstractions.Messages:SynchronizationResponseMessage";

    private final RabbitTemplate qhubTemplate;
    private final ObjectMapper mapper;

    public void send(String correlationId, long transactionId, boolean success,
                     String errorCode, List<String> errorParams) throws IOException {
        if (transactionId <= 0) {
            return;                           // QHUB by ACK nemal ku ktorej transakcii priradiť
        }
        // stabilné messageId: pri opakovaní toho istého ACK rovnaké
        String messageId = UUID.nameUUIDFromBytes(
                ("ack:" + correlationId + ":" + transactionId).getBytes(StandardCharsets.UTF_8)).toString();

        Map<String, Object> result = new LinkedHashMap<>();
        result.put("success", success);
        result.put("errorCode", errorCode);
        result.put("errorParams", errorParams);
        Map<String, Object> envelope = new LinkedHashMap<>();
        envelope.put("messageId", messageId);
        envelope.put("correlationId", correlationId);
        envelope.put("messageType", List.of(MT_RESPONSE));
        envelope.put("sentTime", Instant.now().toString());
        envelope.put("message", Map.of("transactionId", transactionId, "result", result));

        MessageProperties props = new MessageProperties();
        props.setContentType("application/vnd.masstransit+json");   // POVINNÉ
        props.setHeader("MT-MessageType", MT_RESPONSE);
        props.setMessageId(messageId);
        props.setCorrelationId(correlationId);
        props.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
        // ak publikovanie zlyhá, uložte ACK a opakujte, kým neprejde
        qhubTemplate.send("clouderp-sync-response-queue", "",
                new Message(mapper.writeValueAsBytes(envelope), props));
    }
}

public void sendToPeppol(String ublXml, String sender, String receiver,
                         String documentTypeId, String processId) throws IOException {
    Map<String, Object> data = new LinkedHashMap<>();     // PascalCase
    data.put("SenderParticipantId", sender);
    data.put("ReceiverParticipantId", receiver);
    data.put("DocumentTypeId", documentTypeId);
    data.put("ProcessId", processId);
    data.put("UblDocument", ublXml);                          // XML text, nie Base64

    String syncKey = UUID.randomUUID().toString();           // nové pre každý dokument
    Map<String, Object> msg = new LinkedHashMap<>();      // camelCase, bez obálky
    msg.put("transactionId", 0);
    msg.put("eventType", 1);                                  // vždy 1
    msg.put("entityType", 30);
    msg.put("routingKey", null);
    msg.put("ownerTenantId", tenantId);                       // IAM tenant UUID
    msg.put("ownerSystem", systemCode);
    msg.put("deliverySystem", null);
    msg.put("deliveryTenantId", tenantId);
    msg.put("syncKey", syncKey);
    msg.put("schemaVersion", "1.0.0");
    msg.put("environment", environment);
    msg.put("domainSyncVersion", 1);
    msg.put("occurredAt", Instant.now().toString());
    msg.put("data", mapper.writeValueAsString(data));
    msg.put("ignoredProperties", null);

    // outbox: stabilné message_id pre opakované publikovanie, jedinečné correlation_id pre dokument
    OutboxRow row = outbox.save(syncKey, UUID.randomUUID().toString(), mapper.writeValueAsBytes(msg));

    MessageProperties props = new MessageProperties();
    props.setContentType("application/json");                 // NIE masstransit+json
    props.setHeader("MT-MessageType",
            "urn:message:CloudERP.QHubApi.Hub.Abstractions.Messages:SynchronizationMessage");
    props.setMessageId(row.id().toString());
    props.setCorrelationId(row.correlationId());
    props.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
    qhubTemplate.send("clouderp-sync-queue", "", new Message(row.body(), props));
}

using System.Security.Authentication;
using System.Security.Cryptography;
using System.Text;
using System.Text.Json;
using System.Text.Json.Nodes;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

const string ExchangeDelivery = "clouderp-delivery-queue";
const string ExchangeSync = "clouderp-sync-queue";
const string ExchangeSyncResponse = "clouderp-sync-response-queue";
const string MtSync = "urn:message:CloudERP.QHubApi.Hub.Abstractions.Messages:SynchronizationMessage";
const string MtResponse = "urn:message:CloudERP.QHubApi.Hub.Abstractions.Messages:SynchronizationResponseMessage";

var routingKey = $"{systemCode}.{tenantId}";
var delivery = $"{ExchangeDelivery}.{routingKey}";
var uri = new Uri(config["QHub:Url"]!);                      // amqps://messaging.qasida.sk:5671

var factory = new ConnectionFactory
{
    Uri = uri,
    UserName = config["QHub:Username"]!,
    Password = config["QHub:Password"]!,
    AutomaticRecoveryEnabled = true,
    Ssl = new SslOption
    {
        Enabled = true,
        ServerName = uri.Host,
        Version = SslProtocols.Tls12 | SslProtocols.Tls13,
        // CA brokera pridajte do úložiska certifikátov (alebo overte v CertificateValidationCallback);
        // pri mTLS: CertPath = "client.pfx", CertPassphrase = …
    }
};
await using var connection = await factory.CreateConnectionAsync();
await using var channel = await connection.CreateChannelAsync();

// ── Topológia: len vlastné objekty ──
await channel.ExchangeDeclareAsync(delivery, ExchangeType.Fanout, durable: true, autoDelete: false);
await channel.QueueDeclareAsync(delivery, durable: true, exclusive: false, autoDelete: false,
    arguments: new Dictionary<string, object?> { ["x-dead-letter-exchange"] = "clouderp.dlq" });
await channel.QueueBindAsync(delivery, delivery, routingKey: "");
await channel.ExchangeBindAsync(destination: delivery, source: ExchangeDelivery, routingKey: routingKey);
await channel.BasicQosAsync(prefetchSize: 0, prefetchCount: 10, global: false);

// ── Prijímanie: uložiť a potvrdiť (persist-then-process) ──
var consumer = new AsyncEventingBasicConsumer(channel);
consumer.ReceivedAsync += async (_, ea) =>
{
    var envelope = ea.Body.Length > 0 ? JsonNode.Parse(ea.Body.Span) : null;
    var types = envelope?["messageType"]?.ToJsonString() ?? "";
    if (envelope?["message"] is JsonObject msg && !types.Contains("SynchronizationResponseMessage"))
    {
        var transactionId = long.Parse(msg["transactionId"]!.ToString());
        var correlationId = envelope["correlationId"]?.GetValue<string>();
        await inbox.SaveIfNewAsync(transactionId, correlationId, Encoding.UTF8.GetString(ea.Body.Span));
    }
    await channel.BasicAckAsync(ea.DeliveryTag, multiple: false);
};
await channel.BasicConsumeAsync(delivery, autoAck: false, consumer);

// ── Spracovanie uloženej správy (napr. z BackgroundService) ──
async Task ProcessAsync(InboxRow row)
{
    var msg = JsonNode.Parse(row.RawJson)!["message"]!;
    var entityType = msg["entityType"]!.ToString();                  // číslo aj reťazec
    var data = JsonNode.Parse(msg["data"]!.GetValue<string>())!;    // 2. deserializácia, PascalCase
    try
    {
        if (entityType is "30" or "PeppolDocument")
            await ImportDocumentAsync(data);                         // UblDocument = StandardBusinessDocument
        else if (entityType is "32" or "PeppolDocumentEvent")
            await UpdateSentDocumentAsync(data["PeppolDocumentKey"]!["SyncKey"]!.GetValue<string>(),
                                          data["EventType"]!.GetValue<string>(), data);
        await SendAckAsync(row.CorrelationId, row.TransactionId, true, null, null);
    }
    catch (Exception)
    {
        await SendAckAsync(row.CorrelationId, row.TransactionId, false, "ImportFailed",
                           new[] { row.TransactionId.ToString() });
    }
}

// ── Response ACK: MassTransit obálka ──
async Task SendAckAsync(string? correlationId, long transactionId, bool success,
                        string? errorCode, string[]? errorParams)
{
    if (transactionId <= 0) return;
    // stabilné messageId pri opakovaní toho istého ACK
    var messageId = new Guid(MD5.HashData(Encoding.UTF8.GetBytes($"ack:{correlationId}:{transactionId}")));
    var body = JsonSerializer.SerializeToUtf8Bytes(new
    {
        messageId,
        correlationId,
        messageType = new[] { MtResponse },
        sentTime = DateTimeOffset.UtcNow.ToString("O"),
        message = new { transactionId, result = new { success, errorCode, errorParams } }
    });
    var props = new BasicProperties
    {
        ContentType = "application/vnd.masstransit+json",           // POVINNÉ
        MessageId = messageId.ToString(),
        CorrelationId = correlationId,
        DeliveryMode = DeliveryModes.Persistent,
        Headers = new Dictionary<string, object?> { ["MT-MessageType"] = MtResponse }
    };
    await channel.BasicPublishAsync(ExchangeSyncResponse, "", mandatory: false, basicProperties: props, body: body);
}

// ── Odoslanie dokumentu: holá SynchronizationMessage ──
async Task SendToPeppolAsync(OutboxRow row)   // row: uložený dokument s Id, CorrelationId a SyncKey
{
    var data = JsonSerializer.Serialize(new Dictionary<string, string?>   // PascalCase kľúče
    {
        ["SenderParticipantId"] = row.Sender,
        ["ReceiverParticipantId"] = row.Receiver,
        ["DocumentTypeId"] = row.DocumentTypeId,
        ["ProcessId"] = row.ProcessId,
        ["UblDocument"] = row.UblXml                                  // XML text, nie Base64
    });
    var body = JsonSerializer.SerializeToUtf8Bytes(new                 // camelCase, bez obálky
    {
        transactionId = 0,
        eventType = 1,
        entityType = 30,
        routingKey = (string?)null,
        ownerTenantId = tenantId,
        ownerSystem = systemCode,
        deliverySystem = (string?)null,
        deliveryTenantId = tenantId,
        syncKey = row.SyncKey,                                       // nové UUID pre každý dokument
        schemaVersion = "1.0.0",
        environment,
        domainSyncVersion = 1,
        occurredAt = DateTimeOffset.UtcNow.ToString("O"),
        data,
        ignoredProperties = (string[]?)null
    });
    var props = new BasicProperties
    {
        ContentType = "application/json",                            // NIE masstransit+json
        MessageId = row.Id.ToString(),                               // rovnaké pri opakovanom publikovaní
        CorrelationId = row.CorrelationId,                           // jedinečné pre dokument
        DeliveryMode = DeliveryModes.Persistent,
        Headers = new Dictionary<string, object?> { ["MT-MessageType"] = MtSync }
    };
    await channel.BasicPublishAsync(ExchangeSync, "", mandatory: false, basicProperties: props, body: body);
}

import json, os, ssl, uuid
import xml.etree.ElementTree as ET
from datetime import datetime, timezone
import pika

SYSTEM_CODE = os.environ["QHUB_SYSTEM_CODE"]            # napr. MYERP
TENANT_ID = os.environ["QHUB_TENANT_ID"]                # IAM tenant UUID
ROUTING_KEY = f"{SYSTEM_CODE}.{TENANT_ID}"
EXCHANGE_DELIVERY = "clouderp-delivery-queue"
EXCHANGE_SYNC = "clouderp-sync-queue"
EXCHANGE_SYNC_RESPONSE = "clouderp-sync-response-queue"
DELIVERY = f"{EXCHANGE_DELIVERY}.{ROUTING_KEY}"
MT_SYNC = "urn:message:CloudERP.QHubApi.Hub.Abstractions.Messages:SynchronizationMessage"
MT_RESPONSE = "urn:message:CloudERP.QHubApi.Hub.Abstractions.Messages:SynchronizationResponseMessage"
UBL_INVOICE = "{urn:oasis:names:specification:ubl:schema:xsd:Invoice-2}Invoice"

# ── Pripojenie (hodnoty z Registration/Validate, uložené v trezore) ──
params = pika.URLParameters(os.environ["QHUB_URL"])     # amqps://user:pass@messaging.qasida.sk:5671
ctx = ssl.create_default_context(cafile=os.environ["QHUB_CA_CERT"])   # privátna CA brokera
# pri mTLS: ctx.load_cert_chain("client_cert.pem", "client_key.pem")
params.ssl_options = pika.SSLOptions(ctx, params.host)
connection = pika.BlockingConnection(params)
channel = connection.channel()

# ── Topológia: len vlastné objekty, centrálne exchange nedeklarovať ──
channel.exchange_declare(DELIVERY, exchange_type="fanout", durable=True)
channel.queue_declare(DELIVERY, durable=True, arguments={"x-dead-letter-exchange": "clouderp.dlq"})
channel.queue_bind(queue=DELIVERY, exchange=DELIVERY, routing_key="")
channel.exchange_bind(destination=DELIVERY, source=EXCHANGE_DELIVERY, routing_key=ROUTING_KEY)
channel.basic_qos(prefetch_count=10)


def short_id(value):
    # iso6523-actorid-upis::0245:… → 0245:…, busdox-docid-qns::urn:… → urn:…
    if value.startswith(("iso6523-actorid-upis::", "busdox-docid-qns::")):
        return value.split("::", 1)[1]
    return value


def business_document(sbdh_xml):
    # UblDocument prijatého dokumentu je StandardBusinessDocument; hľadajte podľa menného priestoru
    root = ET.fromstring(sbdh_xml.encode("utf-8"))
    return next(root.iter(UBL_INVOICE), None)


def send_ack(correlation_id, transaction_id, success, error_code=None, error_params=None):
    if transaction_id <= 0:
        return
    message_id = str(uuid.uuid5(uuid.NAMESPACE_URL, f"ack:{correlation_id}:{transaction_id}"))  # stabilné
    envelope = {
        "messageId": message_id,
        "correlationId": correlation_id,
        "messageType": [MT_RESPONSE],
        "sentTime": datetime.now(timezone.utc).isoformat(),
        "message": {"transactionId": transaction_id,
                    "result": {"success": success, "errorCode": error_code, "errorParams": error_params}},
    }
    channel.basic_publish(
        exchange=EXCHANGE_SYNC_RESPONSE, routing_key="",
        body=json.dumps(envelope).encode("utf-8"),
        properties=pika.BasicProperties(
            content_type="application/vnd.masstransit+json",   # POVINNÉ
            headers={"MT-MessageType": MT_RESPONSE},
            message_id=message_id, correlation_id=correlation_id, delivery_mode=2))


def on_message(ch, method, props, body):
    envelope = json.loads(body) if body else {}
    msg = envelope.get("message")
    if msg is None or any("SynchronizationResponseMessage" in t for t in envelope.get("messageType", [])):
        ch.basic_ack(method.delivery_tag)                   # prázdna správa alebo ACK obálka
        return
    correlation_id = envelope.get("correlationId")
    transaction_id = int(msg.get("transactionId") or 0)
    if not save_if_new(transaction_id, correlation_id, body):   # vaša tabuľka s unikátnymi indexmi
        ch.basic_ack(method.delivery_tag)                   # duplikát
        return
    try:
        entity_type = str(msg["entityType"])                # číslo aj reťazec
        data = json.loads(msg["data"])                      # 2. deserializácia, PascalCase
        if entity_type in ("30", "PeppolDocument"):
            import_document(short_id(data["SenderParticipantId"]), business_document(data["UblDocument"]))
        elif entity_type in ("32", "PeppolDocumentEvent"):
            update_sent_document(data["PeppolDocumentKey"]["SyncKey"], data["EventType"], data)
        send_ack(correlation_id, transaction_id, True)
    except Exception:
        send_ack(correlation_id, transaction_id, False, "ImportFailed", [str(transaction_id)])
    ch.basic_ack(method.delivery_tag)


def send_to_peppol(ubl_xml, sender, receiver, document_type_id, process_id,
                   outbox_id, correlation_id, sync_key):
    data = {"SenderParticipantId": sender, "ReceiverParticipantId": receiver,   # PascalCase
            "DocumentTypeId": document_type_id, "ProcessId": process_id, "UblDocument": ubl_xml}
    message = {                                             # camelCase, bez obálky
        "transactionId": 0, "eventType": 1, "entityType": 30, "routingKey": None,
        "ownerTenantId": TENANT_ID, "ownerSystem": SYSTEM_CODE, "deliverySystem": None,
        "deliveryTenantId": TENANT_ID, "syncKey": sync_key, "schemaVersion": "1.0.0",
        "environment": os.environ["QHUB_ENVIRONMENT"], "domainSyncVersion": 1,
        "occurredAt": datetime.now(timezone.utc).isoformat(),
        "data": json.dumps(data), "ignoredProperties": None}
    channel.basic_publish(
        exchange=EXCHANGE_SYNC, routing_key="",
        body=json.dumps(message).encode("utf-8"),
        properties=pika.BasicProperties(
            content_type="application/json",                # NIE masstransit+json
            headers={"MT-MessageType": MT_SYNC},
            message_id=outbox_id,                           # rovnaké pri opakovanom publikovaní
            correlation_id=correlation_id,                  # jedinečné pre dokument
            delivery_mode=2))


channel.basic_consume(queue=DELIVERY, on_message_callback=on_message)
channel.start_consuming()