/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.beam.sdk.io.solace.data;

import com.google.auto.value.AutoValue;
import com.solacesystems.jcsmp.BytesXMLMessage;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import org.apache.beam.sdk.schemas.AutoValueSchema;
import org.apache.beam.sdk.schemas.annotations.DefaultSchema;
import org.checkerframework.checker.nullness.qual.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Provides core data models and utilities for working with Solace messages in the context of Apache
 * Beam pipelines. This class includes representations for Solace topics, queues, destinations, and
 * message records, as well as a utility for converting Solace messages into Beam-compatible
 * records.
 */
public class Solace {
  /** Represents a Solace queue. */
  public static class Queue {
    private final String name;

    private Queue(String name) {
      this.name = name;
    }

    public static Queue fromName(String name) {
      return new Queue(name);
    }

    public String getName() {
      return name;
    }
  }
  /** Represents a Solace topic. */
  public static class Topic {
    private final String name;

    private Topic(String name) {
      this.name = name;
    }

    public static Topic fromName(String name) {
      return new Topic(name);
    }

    public String getName() {
      return name;
    }
  }
  /** Represents a Solace destination type. */
  public enum DestinationType {
    TOPIC,
    QUEUE,
    UNKNOWN
  }

  /** Represents a Solace message destination (either a Topic or a Queue). */
  @AutoValue
  @DefaultSchema(AutoValueSchema.class)
  public abstract static class Destination {
    /**
     * Gets the name of the destination.
     *
     * @return The destination name.
     */
    public abstract String getName();

    /**
     * Gets the type of the destination (TOPIC, QUEUE or UNKNOWN).
     *
     * @return The destination type.
     */
    public abstract DestinationType getType();

    static Builder builder() {
      return new AutoValue_Solace_Destination.Builder();
    }

    @AutoValue.Builder
    abstract static class Builder {
      abstract Builder setName(String name);

      abstract Builder setType(DestinationType type);

      abstract Destination build();
    }
  }

  /** Represents a Solace message record with its associated metadata. */
  @AutoValue
  @DefaultSchema(AutoValueSchema.class)
  public abstract static class Record {
    /**
     * Gets the unique identifier of the message, a string for an application-specific message
     * identifier.
     *
     * <p>Mapped from {@link BytesXMLMessage#getApplicationMessageId()}
     *
     * @return The message ID, or null if not available.
     */
    public abstract @Nullable String getMessageId();

    /**
     * Gets the payload of the message as a ByteString.
     *
     * <p>Mapped from {@link BytesXMLMessage#getBytes()}
     *
     * @return The message payload.
     */
    public abstract ByteBuffer getPayload();
    /**
     * Gets the destination (topic or queue) to which the message was sent.
     *
     * <p>Mapped from {@link BytesXMLMessage#getDestination()}
     *
     * @return The destination, or null if not available.
     */
    public abstract @Nullable Destination getDestination();

    /**
     * Gets the message expiration time in milliseconds since the Unix epoch.
     *
     * <p>A value of 0 indicates the message does not expire.
     *
     * <p>Mapped from {@link BytesXMLMessage#getExpiration()}
     *
     * @return The expiration timestamp.
     */
    public abstract long getExpiration();

    /**
     * Gets the priority level of the message (0-255, higher is more important). -1 if not set.
     *
     * <p>Mapped from {@link BytesXMLMessage#getPriority()}
     *
     * @return The message priority.
     */
    public abstract int getPriority();

    /**
     * Indicates whether the message has been redelivered due to a prior delivery failure.
     *
     * <p>Mapped from {@link BytesXMLMessage#getRedelivered()}
     *
     * @return True if redelivered, false otherwise.
     */
    public abstract boolean getRedelivered();

    /**
     * Gets the destination to which replies to this message should be sent.
     *
     * <p>Mapped from {@link BytesXMLMessage#getReplyTo()}
     *
     * @return The reply-to destination, or null if not specified.
     */
    public abstract @Nullable Destination getReplyTo();

    /**
     * Gets the timestamp (in milliseconds since the Unix epoch) when the message was received by
     * the Solace broker.
     *
     * <p>Mapped from {@link BytesXMLMessage#getReceiveTimestamp()}
     *
     * @return The timestamp.
     */
    public abstract long getReceiveTimestamp();

    /**
     * Gets the timestamp (in milliseconds since the Unix epoch) when the message was sent by the
     * sender. Can be null if not provided.
     *
     * @return The sender timestamp, or null if not available.
     */
    public abstract @Nullable Long getSenderTimestamp();

    /**
     * Gets the sequence number of the message (if applicable).
     *
     * <p>Mapped from {@link BytesXMLMessage#getSequenceNumber()}
     *
     * @return The sequence number, or null if not available.
     */
    public abstract @Nullable Long getSequenceNumber();

    /**
     * The number of milliseconds before the message is discarded or moved to Dead Message Queue. A
     * value of 0 means the message will never expire. The default value is 0.
     *
     * <p>Mapped from {@link BytesXMLMessage#getTimeToLive()}
     *
     * @return The time-to-live value.
     */
    public abstract long getTimeToLive();

    /**
     * Gets the ID for the message within its replication group (if applicable).
     *
     * <p>Mapped from {@link BytesXMLMessage#getReplicationGroupMessageId()}
     *
     * <p>The ID for a particular message is only guaranteed to be the same for a particular copy of
     * a message on a particular queue or topic endpoint within a replication group. The same
     * message on different queues or topic endpoints within the same replication group may or may
     * not have the same replication group message ID. See more at <a
     * href="https://docs.solace.com/API/API-Developer-Guide/Detecting-Duplicate-Mess.htm">https://docs.solace.com/API/API-Developer-Guide/Detecting-Duplicate-Mess.htm</a>
     *
     * @return The replication group message ID, or null if not present.
     */
    public abstract @Nullable String getReplicationGroupMessageId();
    /**
     * Gets the attachment data of the message as a ByteString, if any. This might represent files
     * or other binary content associated with the message.
     *
     * <p>Mapped from {@link BytesXMLMessage#getAttachmentByteBuffer()}
     *
     * @return The attachment data, or an empty ByteString if no attachment is present.
     */
    public abstract ByteBuffer getAttachmentBytes();

    static Builder builder() {
      return new AutoValue_Solace_Record.Builder();
    }

    @AutoValue.Builder
    abstract static class Builder {
      abstract Builder setMessageId(@Nullable String messageId);

      abstract Builder setPayload(ByteBuffer payload);

      abstract Builder setDestination(@Nullable Destination destination);

      abstract Builder setExpiration(long expiration);

      abstract Builder setPriority(int priority);

      abstract Builder setRedelivered(boolean redelivered);

      abstract Builder setReplyTo(@Nullable Destination replyTo);

      abstract Builder setReceiveTimestamp(long receiveTimestamp);

      abstract Builder setSenderTimestamp(@Nullable Long senderTimestamp);

      abstract Builder setSequenceNumber(@Nullable Long sequenceNumber);

      abstract Builder setTimeToLive(long timeToLive);

      abstract Builder setReplicationGroupMessageId(@Nullable String replicationGroupMessageId);

      abstract Builder setAttachmentBytes(ByteBuffer attachmentBytes);

      abstract Record build();
    }
  }
  /**
   * A utility class for mapping {@link BytesXMLMessage} instances to {@link Solace.Record} objects.
   * This simplifies the process of converting raw Solace messages into a format suitable for use
   * within Apache Beam pipelines.
   */
  public static class SolaceRecordMapper {
    private static final Logger LOG = LoggerFactory.getLogger(SolaceRecordMapper.class);
    /**
     * Maps a {@link BytesXMLMessage} (if not null) to a {@link Solace.Record}.
     *
     * <p>Extracts relevant information from the message, including payload, metadata, and
     * destination details.
     *
     * @param msg The Solace message to map.
     * @return A Solace Record representing the message, or null if the input message was null.
     */
    public static @Nullable Record map(@Nullable BytesXMLMessage msg) {
      if (msg == null) {
        return null;
      }

      ByteArrayOutputStream payloadBytesStream = new ByteArrayOutputStream();
      if (msg.getContentLength() != 0) {
        try {
          payloadBytesStream.write(msg.getBytes());
        } catch (IOException e) {
          LOG.error("Could not write bytes from the BytesXMLMessage to the Solace.record.", e);
        }
      }

      ByteArrayOutputStream attachmentBytesStream = new ByteArrayOutputStream();
      if (msg.getAttachmentContentLength() != 0) {
        try {
          attachmentBytesStream.write(msg.getAttachmentByteBuffer().array());
        } catch (IOException e) {
          LOG.error(
              "Could not AttachmentByteBuffer from the BytesXMLMessage to the Solace.record.", e);
        }
      }

      Destination replyTo = getDestination(msg.getCorrelationId(), msg.getReplyTo());
      Destination destination = getDestination(msg.getCorrelationId(), msg.getDestination());
      return Record.builder()
          .setMessageId(msg.getApplicationMessageId())
          .setPayload(ByteBuffer.wrap(payloadBytesStream.toByteArray()))
          .setDestination(destination)
          .setExpiration(msg.getExpiration())
          .setPriority(msg.getPriority())
          .setRedelivered(msg.getRedelivered())
          .setReplyTo(replyTo)
          .setReceiveTimestamp(msg.getReceiveTimestamp())
          .setSenderTimestamp(msg.getSenderTimestamp())
          .setSequenceNumber(msg.getSequenceNumber())
          .setTimeToLive(msg.getTimeToLive())
          .setReplicationGroupMessageId(
              msg.getReplicationGroupMessageId() != null
                  ? msg.getReplicationGroupMessageId().toString()
                  : null)
          .setAttachmentBytes(ByteBuffer.wrap(attachmentBytesStream.toByteArray()))
          .build();
    }

    private static @Nullable Destination getDestination(
        String msgId, com.solacesystems.jcsmp.Destination originalDestinationField) {
      if (originalDestinationField == null) {
        return null;
      }
      Destination.Builder destinationBuilder =
          Destination.builder().setName(originalDestinationField.getName());
      if (originalDestinationField instanceof com.solacesystems.jcsmp.Topic) {
        destinationBuilder.setType(DestinationType.TOPIC);
      } else if (originalDestinationField instanceof com.solacesystems.jcsmp.Queue) {
        destinationBuilder.setType(DestinationType.QUEUE);
      } else {
        LOG.error(
            "SolaceIO: Unknown destination type type for message {}, setting to {}",
            msgId,
            DestinationType.UNKNOWN.name());
        destinationBuilder.setType(DestinationType.UNKNOWN);
      }
      return destinationBuilder.build();
    }
  }
}
