package com.atlassian.messagequeue.internal.core;
   
   import com.atlassian.messagequeue.Message;
   import com.atlassian.messagequeue.MessageInformationService;
   import com.atlassian.messagequeue.MessageRunnerKey;
   import com.atlassian.messagequeue.TenantDataIdSupplier;
   import com.atlassian.tenant.api.TenantContext;
   import com.atlassian.tenant.api.TenantContextProvider;
   
  import static com.atlassian.messagequeue.internal.core.NestedMessageConstants.MESSAGE_RUNNER_KEY_ATTRIBUTE_NAME;
  import static com.atlassian.messagequeue.internal.core.NestedMessageConstants.TENANT_DATA_ID_ATTRIBUTE_NAME;
  import static com.atlassian.messagequeue.internal.core.NestedMessageConstants.TENANT_ID_ATTRIBUTE_NAME;
  import static java.util.Objects.requireNonNull;
  
  /**
   * Message information service.
   */
  public class DefaultMessageInformationService implements MessageInformationService {
      private final TenantContextProvider tenantContextProvider;
      private final String queueUrl;
      private final NestedMessageSerializer nestedMessageSerializer;
      private final TenantDataIdSupplier tenantDataIdSupplier;
  
      /**
       * @param queueUrl The URL that atlassian-messagequeue will be producing to and consuming from.
       * @param tenantContextProvider the tenant context provider
       * @param nestedMessageSerializer the nested message serializer
       * @param tenantDataIdSupplier the tenant data id supplier
       */
      public DefaultMessageInformationService(String queueUrl,
                                              TenantContextProvider tenantContextProvider,
                                              NestedMessageSerializer nestedMessageSerializer,
                                              TenantDataIdSupplier tenantDataIdSupplier) {
          this.queueUrl = requireNonNull(queueUrl, "queueUrl");
          this.tenantContextProvider = requireNonNull(tenantContextProvider, "tenantContextProvider");
          this.nestedMessageSerializer = requireNonNull(nestedMessageSerializer, "nestedMessageSerializer");
          this.tenantDataIdSupplier = requireNonNull(tenantDataIdSupplier);
      }
  
      @Override
      public String getQueueUrl(MessageRunnerKey messageRunnerKey) {
          return queueUrl;
      }
  
      @Override
      public String toPayload(Message message) {
          final NestedMessage nestedMessage = new NestedMessage();
  
          final TenantContext tenantContext = tenantContextProvider.getTenantContext();
          if (tenantContext == null) {
              throw new IllegalStateException("No tenant context available from which to derive a tenant ID");
          } else if (tenantContext.getTenantId() == null || "".equals(tenantContext.getTenantId())) {
              throw new IllegalStateException("Null or empty tenantId found in tenant context");
          }
  
          nestedMessage
                  .addAttribute(TENANT_ID_ATTRIBUTE_NAME, requireNonNull(tenantContext.getTenantId(), "tenantId"))
                  .addAttribute(MESSAGE_RUNNER_KEY_ATTRIBUTE_NAME, requireNonNull(message.getRunnerKey(), "messageRunnerKey").toString())
                  .addAttribute(TENANT_DATA_ID_ATTRIBUTE_NAME, tenantDataIdSupplier.get())
                  .setPayload(message.getPayload().orElse(null));
  
          return nestedMessageSerializer.serialize(nestedMessage);
      }
  
      @Override
      public Message fromPayload(String payload) {
          if (payload == null || "".equals(payload)) {
              throw new IllegalArgumentException("payload not be null or empty");
          }
  
          final NestedMessage nestedMessage = nestedMessageSerializer.deserialize(payload);
          final String messageRunnerKey = nestedMessage.getAttribute(MESSAGE_RUNNER_KEY_ATTRIBUTE_NAME);
  
          return Message.create(MessageRunnerKey.of(messageRunnerKey), nestedMessage.getPayload());
      }
  }