/*
 * JORAM: Java(TM) Open Reliable Asynchronous Messaging
 * Copyright (C) 2017 - 2025 ScalAgent Distributed Technologies
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307
 * USA.
 *
 * Initial developer(s): ScalAgent Distributed Technologies
 * Contributor(s): 
 */
package com.scalagent.joram.mom.dest.rest;

import java.io.UnsupportedEncodingException;
import java.net.URI;
import java.net.URLEncoder;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.TimeUnit;
import java.util.Properties;

import jakarta.ws.rs.client.Client;
import jakarta.ws.rs.client.ClientBuilder;
import jakarta.ws.rs.client.Entity;
import jakarta.ws.rs.client.WebTarget;
import jakarta.ws.rs.core.Form;
import jakarta.ws.rs.core.MediaType;
import jakarta.ws.rs.core.Response;
import jakarta.ws.rs.core.UriBuilder;
import jakarta.ws.rs.ext.RuntimeDelegate;

import org.objectweb.joram.mom.dest.DistributionHandler;
import org.objectweb.joram.shared.DestinationConstants;
import org.objectweb.joram.shared.messages.Message;
import org.objectweb.util.monolog.api.BasicLevel;
import org.objectweb.util.monolog.api.Logger;

import com.google.gson.Gson;
import com.google.gson.GsonBuilder;

import fr.dyade.aaa.common.Debug;

/**
 * Distribution handler for the REST distribution bridge.
 */
public final class RESTDistribution implements DistributionHandler {
  private static final Logger logger = Debug.getLogger(RESTDistribution.class.getName());
  private static final boolean DEBUG = logger.isLoggable(BasicLevel.DEBUG);

  static {
    System.setProperty(RuntimeDelegate.JAXRS_RUNTIME_DELEGATE_PROPERTY, "org.glassfish.jersey.internal.RuntimeDelegateImpl");
    System.setProperty(ClientBuilder.JAXRS_DEFAULT_CLIENT_BUILDER_PROPERTY, "org.glassfish.jersey.client.JerseyClientBuilder");
  }
  
  private Properties properties = null;
  
  private String hostName = "localhost";
  private int port = 8989;

  // If true, use the API with authentication parameters in URL.
  private boolean useOldAPI = false;
  
  private int connectTimeout = 5000;
  // Be careful, we need to add timeout for the client configuration.
  private int readTimeout = 5000;
  
  private String userName = "anonymous";
  private String password = "anonymous";

  private String destName = null;
  
  private String idleTimeout = "60";

  // TODO (AF): Currently, prodName and clientId are always null.
  private String prodName = null;
  private String clientId = null;

  private Client client;
  
  private URI uriSendNextMsg = null;
  private URI uriCloseProducer = null;
  
  @Override
  public void init(Properties properties, boolean firstTime) {
    this.properties = properties;
    
    // Get URI informations: host, port, etc.
    if (properties.containsKey(DestinationConstants.REST_HOST_PROP)) {
      hostName = properties.getProperty(DestinationConstants.REST_HOST_PROP);
    } else {
      logger.log(BasicLevel.WARN,
          "Missing property " + DestinationConstants.REST_HOST_PROP + ", use default value: " + hostName);
    }
    if (properties.containsKey(DestinationConstants.REST_PORT_PROP)) {
      try {
        port = Integer.parseInt(properties.getProperty(DestinationConstants.REST_PORT_PROP));
      } catch (NumberFormatException exc) {
        logger.log(BasicLevel.ERROR,
                   "Property " + DestinationConstants.REST_PORT_PROP + '=' + properties.getProperty(DestinationConstants.REST_PORT_PROP) +
                   " could not be parsed properly, use default value: " + port);
      }
    } else {
      logger.log(BasicLevel.WARN,
          "Missing property " + DestinationConstants.REST_PORT_PROP + ", use default value: " + port);
    }
    
    if (properties.containsKey(DestinationConstants.REST_USE_OLD_API)) {
      useOldAPI = Boolean.parseBoolean(properties.getProperty(DestinationConstants.REST_USE_OLD_API));
      logger.log(BasicLevel.INFO, DestinationConstants.REST_USE_OLD_API + "=" + useOldAPI);
    }

    // Get authentication informations: user, password.
    if (properties.containsKey(DestinationConstants.REST_USERNAME_PROP)) {
      userName = properties.getProperty(DestinationConstants.REST_USERNAME_PROP);
    } else {
      logger.log(BasicLevel.WARN,
          "Missing property " + DestinationConstants.REST_USERNAME_PROP + ", use default value: " + userName);
    }
    if (properties.containsKey(DestinationConstants.REST_PASSWORD_PROP)) {
      password = properties.getProperty(DestinationConstants.REST_PASSWORD_PROP);
    } else {
      logger.log(BasicLevel.WARN,
          "Missing property " + DestinationConstants.REST_PASSWORD_PROP + ", use default value: " + password);
    }

    if (properties.containsKey(DestinationConstants.IDLETIMEOUT_PROP)) {
      try {
        idleTimeout = properties.getProperty(DestinationConstants.IDLETIMEOUT_PROP);
      } catch (NumberFormatException exc) { }
    }

    if (properties.containsKey(DestinationConstants.REST_CONNECT_TIMEOUT_PROP)) {
      try {
        connectTimeout = Integer.parseInt(properties.getProperty(DestinationConstants.REST_CONNECT_TIMEOUT_PROP));
      } catch (NumberFormatException exc) {
        logger.log(BasicLevel.ERROR,
            "Property " + DestinationConstants.REST_CONNECT_TIMEOUT_PROP + '=' + properties.getProperty(DestinationConstants.REST_CONNECT_TIMEOUT_PROP) + 
            " could not be parsed properly, use default value: " + connectTimeout);
      }
    } else {
      logger.log(BasicLevel.WARN,
          "Missing property " + DestinationConstants.REST_CONNECT_TIMEOUT_PROP + ", use default value: " + connectTimeout);
    }

    if (properties.containsKey(DestinationConstants.REST_READ_TIMEOUT_PROP)) {
      try {
        readTimeout = Integer.parseInt(properties.getProperty(DestinationConstants.REST_READ_TIMEOUT_PROP));
      } catch (NumberFormatException exc) {
        logger.log(BasicLevel.ERROR,
            "Property " + DestinationConstants.REST_READ_TIMEOUT_PROP + '=' + properties.getProperty(DestinationConstants.REST_READ_TIMEOUT_PROP) + 
            " could not be parsed properly, use default value: " + readTimeout);
      }
    } else {
      logger.log(BasicLevel.WARN,
          "Missing property " + DestinationConstants.REST_READ_TIMEOUT_PROP + ", use default value: " + readTimeout);
    }

    // Get Destination
    destName = properties.getProperty(DestinationConstants.DESTINATION_NAME_PROP);
    if (destName == null) {
      logger.log(BasicLevel.ERROR,
          "Missing Destination JNDI name, should fixed property " + DestinationConstants.DESTINATION_NAME_PROP);
    } else {
      try {
        destName = URLEncoder.encode(destName, "UTF-8");
      } catch (UnsupportedEncodingException exc) {
        // Should never happened
      }
    }
  }
  
  private void createProducer(String userName, String password) throws Exception {
    if (uriCloseProducer != null)
      return;

    Response response1 = null, response2 = null;
    try {
      URI base = UriBuilder.fromUri("http://" + hostName + ":" + port + "/joram/").build();
      if (DEBUG)
        logger.log(BasicLevel.DEBUG,
                   "RestDistribution.createProducer(), use base URI " + base);

      // Initializes URI
      uriSendNextMsg = null;

      // Initializes Rest client and target
      try {
        if (client != null) client.close();
        // It seems that there is no exceptions thrown by these methods.
        client = ClientBuilder.newBuilder()
            .connectTimeout(connectTimeout, TimeUnit.MILLISECONDS).readTimeout(readTimeout, TimeUnit.MILLISECONDS).build();
        for (Object key : properties.keySet()) {
          try {
            if (key instanceof String)
              client.property((String) key, properties.get(key));
          } catch (Exception e) {
            logger.log(BasicLevel.ERROR,
                       "RestAcquisitionAsync.createConsumer(): cannot set client property " + key + "=" + properties.get(key));
          }
        }
      } catch (Exception exc) {
        if (DEBUG)
          logger.log(BasicLevel.ERROR,
                     "RestDistribution.createProducer(): cannot initialize Rest client", exc);
        else
          logger.log(BasicLevel.ERROR,
                     "RestDistribution.createProducer(): cannot initialize Rest client - " + exc.getMessage());
        throw exc;
      }

      // Get the destination
      try {
        response1 = client.target(base).path("jndi").path(destName)
            .request().accept(MediaType.TEXT_PLAIN).head();
      } catch (Exception exc) {
        if (DEBUG)
          logger.log(BasicLevel.ERROR,
                     "RestDistribution.createProducer(): cannot get destination " + destName, exc);
        else
          logger.log(BasicLevel.ERROR,
                     "RestDistribution.createProducer(): cannot get destination " + destName + "\n\t" + exc.getMessage());
        throw exc;
      }

      if (response1.getStatus() == 201) {
        if (DEBUG)
          logger.log(BasicLevel.DEBUG,
                     "RestDistribution.createProducer(): get destination -> " + response1.getStatusInfo());
      } else {
        logger.log(BasicLevel.ERROR, "RestDistribution.createProducer(): cannot get destination " + destName + " -> " + response1.getStatusInfo());
        throw new Exception("Cannot get destination " + destName);
      }

      try {
        WebTarget target;

        /* This code corresponds to the last version of the Rest/JMS API URI. Currently it can not be
         * used for compatibility reason.
         * 
         * uriCreateProducer = response.getLink("create-producer-fp").getUri();
         * if (DEBUG)
         *   logger.log(BasicLevel.DEBUG, "RestDistribution.createProducer(): create-producer = " + uriCreateProducer);
         * // TODO (AF): We should fix name and client-id to be unique. target =
         * client.target(uriCreateProducer);
         */

        if (useOldAPI) {
          URI uriCreateProducer = response1.getLink("create-producer").getUri();
          if (DEBUG)
            logger.log(BasicLevel.DEBUG,
                       "RestDistribution.createProducer(): create-producer = " + uriCreateProducer);

          // TODO (AF): We should fix name and client-id to be unique.
          target = client.target(uriCreateProducer);
        } else {
          target = client.target(base).path("jndi").path(destName).path(org.objectweb.joram.tools.rest.jms.JmsService.JMS_CREATE_PROD_FP);
        }
        if (DEBUG)
          logger.log(BasicLevel.DEBUG, "RestDistribution.createProducer(): create-consumer = " + target.getUri());

        if (prodName != null)  target = target.queryParam("name", prodName);
        if (clientId != null)  target = target.queryParam("client-id", clientId);

        Form auth = new Form();
        if (useOldAPI) {
          if (userName != null) target = target.queryParam("user", userName);
          if (password != null) target = target.queryParam("password", password);
        } else {
          if (userName != null) auth.param("user", userName);
          if (password != null) auth.param("password", password);
        }

        if (idleTimeout != null)  target = target.queryParam("idle-timeout", idleTimeout);

        if (useOldAPI) {
          response2 = target.request().accept(MediaType.TEXT_PLAIN).post(null);
        } else {
          response2 = target.request().accept(MediaType.TEXT_PLAIN).post(Entity.entity(auth, MediaType.APPLICATION_FORM_URLENCODED));
        }
      } catch (Exception exc) {
        if (DEBUG)
          logger.log(BasicLevel.ERROR,
                     "RestDistribution.createProducer(): cannot create producer", exc);
        else
          logger.log(BasicLevel.ERROR,
                     "RestDistribution.createProducer(): cannot create producer - " + exc.getMessage());

        throw exc;
      }

      if (response2.getStatus() == 201) {
        if (DEBUG)
          logger.log(BasicLevel.DEBUG,
                     "RestDistribution.createProducer(): create producer -> " + response2.getStatusInfo());

        uriCloseProducer = response2.getLink("close-context").getUri();
        uriSendNextMsg = response2.getLink("send-next-message").getUri();
      } else {
        logger.log(BasicLevel.ERROR, "RestDistribution.createProducer(): cannot create producer -> " + response2.getStatusInfo());
        throw new Exception("Cannot create producer");
      }
    } finally {
      if (response1 != null) response1.close();
      if (response2 != null) response2.close();

      if (uriCloseProducer == null) {
        // the createProducer has failed, closes the client if needed.
        if (client != null) client.close();
        client = null;
      }
    }
  }

  @Override
  public void distribute(Message fullMessage) throws Exception {
  	if (DEBUG)
      logger.log(BasicLevel.DEBUG, "RESTDistribution.distribute(" + fullMessage + ')');
    
  	if (destName == null) {
  	  logger.log(BasicLevel.ERROR,
  	      "Missing Destination JNDI name, should fixed property " + DestinationConstants.DESTINATION_NAME_PROP);
  	  throw new Exception("Missing Destination JNDI name, should fixed property " + DestinationConstants.DESTINATION_NAME_PROP);
  	}

  	try {
  	  createProducer(userName, password);
    } catch (Exception exc) {
      throw exc;
    }
  	
  	try {
      sendNextMessage(fullMessage);
    } catch (Exception e) {
      closeProducer();
      throw e;
    }
  }
  
  
  private Map<String, Object> getMapMessageToJsonBodyMap(Message fullMessage) throws MessageFormatException {
    Map msgMap = null;
    
    if (fullMessage.isNullBody())
      throw new MessageFormatException("Message body is null: " + fullMessage.id);
    
    try {
      msgMap = fullMessage.getMap();
    } catch (Exception exc) {
      MessageFormatException jE = new MessageFormatException("Cannot get message body: " + fullMessage.id);
      jE.setLinkedException(exc);
      throw jE;
    }
    if (msgMap == null) return null;
    
    HashMap<String, Object> jsonBodyMap = new HashMap<>();
    Iterator<Map.Entry> entries = msgMap.entrySet().iterator();
    while (entries.hasNext()) {
      Entry entry = entries.next();
      String key = (String) entry.getKey();
      Object v = entry.getValue();
      if (v != null) {
        String[] value = new String[2];
        if (v instanceof byte[]) {
          try {
            value[0] = new String((byte[])v, "UTF-8");
          } catch (UnsupportedEncodingException e) {
            if (logger.isLoggable(BasicLevel.WARN))
              logger.log(BasicLevel.WARN, "", e);
            MessageFormatException jE = new MessageFormatException("Error while encode the bytes to string.");
            jE.setLinkedException(e);
            throw jE;
          }
        } else {
          value[0] = ""+v;
        }
        value[1] = v.getClass().getName();
        jsonBodyMap.put(key, value);
      }
    }
    return jsonBodyMap;
  }
  
  public static final String BytesMessage = "BytesMessage";
  public static final String MapMessage = "MapMessage";
  public static final String TextMessage = "TextMessage";

  // TODO (AF): May be we should use JSon transformation from 'shared message'.
  private void sendNextMessage(Message fullMessage) throws Exception {
    if (uriSendNextMsg == null) {
      logger.log(BasicLevel.ERROR, "RestDistribution.sendNextMessage(): URI not initialized.");
      throw new Exception("URI not initialized");
    }
    
    Response response = null;
    try {
      // TODO (AF): May be we should always send JSon messages.
      if (fullMessage.properties == null && fullMessage.type == Message.TEXT) {
        // If the message contains only text and defines no properties, we can send an
        // optimized message without using JSON.
        
        if (DEBUG)
          logger.log(BasicLevel.DEBUG,
              "RestDistribution.sendNextMessage: Send simple message");

        WebTarget target = client.target(uriSendNextMsg)
            .queryParam("delivery-mode", (fullMessage.persistent?Message.PERSISTENT:Message.NON_PERSISTENT))
            .queryParam("priority", fullMessage.priority);
        if (fullMessage.deliveryTime > 0)
          target = target.queryParam("delivery-time", fullMessage.deliveryTime);
        if (fullMessage.expiration > 0)
          target = target.queryParam("time-to-live", (fullMessage.expiration - System.currentTimeMillis()));
        if (fullMessage.correlationId != null && !fullMessage.correlationId.isEmpty())
          target = target.queryParam("correlation-id", fullMessage.correlationId);

        response =  target.request().accept(MediaType.TEXT_PLAIN).post(Entity.entity(fullMessage.getText(), MediaType.TEXT_PLAIN));

        if (response.getStatus() != 200) {
          logger.log(BasicLevel.ERROR, "RestDistribution.sendNextMessage: cannot send message -> " + response.getStatus());
          throw new Exception("Cannot send message: " + response.getStatus());
        } else {
          if (DEBUG)
            logger.log(BasicLevel.DEBUG,
                       "RestDistribution.sendNextMessage: message sent -> " + response.getStatus());
        }
      } else {
        if (DEBUG)
          logger.log(BasicLevel.DEBUG,
              "RestDistribution.sendNextMessage: Send JSON message");

        // Use JSON to encode the message
        HashMap<String, Object> maps = new HashMap<>();

        // Encode message body
        if (fullMessage.type == Message.TEXT) {
          maps.put("type", TextMessage);
          maps.put("body", fullMessage.getText());
        } else if (fullMessage.type == Message.BYTES) {
          maps.put("type", BytesMessage);
          maps.put("body", fullMessage.getBody());
        } else if (fullMessage.type == Message.MAP) {
          maps.put("type", MapMessage);
          maps.put("body", getMapMessageToJsonBodyMap(fullMessage));
        } else {
          logger.log(BasicLevel.ERROR, "RestDistribution.sendNextMessage: type " + fullMessage.type + " not yet supported.");
          throw new Exception("Message type " + fullMessage.type + " not yet supported.");
        }

        // Transform JMS message properties
        if (fullMessage.properties != null) {
          HashMap<String, Object> props = new HashMap<>();
          Enumeration e = fullMessage.properties.keys();
          while (e.hasMoreElements()) {
            String key = (String) e.nextElement();
            Object value = fullMessage.properties.get(key);
            props.put(key, new String[]{value.toString(), value.getClass().getName()});
          }
          maps.put("properties", props);
        }

        // message header
        HashMap<String, Object> header = new HashMap<>();
        if (fullMessage.correlationId != null)
          header.put("CorrelationID", new String[]{fullMessage.correlationId, String.class.getName()});
        if (fullMessage.priority > 0)
          header.put("Priority", new String[]{""+fullMessage.priority, Integer.class.getName()});
        // TODO (AF): not correct?
        //      header.put("Type", message.jmsType);

        //TODO (AF): other headers prop
        maps.put("header", header);

        Gson gson = new GsonBuilder().create();
        String json = gson.toJson(maps);

        if (DEBUG)
          logger.log(BasicLevel.DEBUG, "RESTDistribution.sendMessage: json " + json);

        // Send next message
        response = client.target(uriSendNextMsg).request().accept(MediaType.TEXT_PLAIN).post(Entity.entity(json, MediaType.APPLICATION_JSON));

        if (response.getStatus() != 200) {
          logger.log(BasicLevel.ERROR, "RestDistribution.sendNextMessage: cannot send message -> " + response.getStatus());
          throw new Exception("Cannot send message: " + response.getStatus());
        } else {
          if (DEBUG)
            logger.log(BasicLevel.DEBUG,
                       "RestDistribution.sendNextMessage: message sent -> " + response.getStatus());
        }
      }

      uriSendNextMsg = response.getLink("send-next-message").getUri();
    } finally {
      if (response != null) response.close();
    }
  }
  
  /**
   * Close the producer.
   */
  private void closeProducer() {
    Response response = null;
    try {
      if (uriCloseProducer == null) return;

      // Try to close the producer on the other side.
      response = client.target(uriCloseProducer).request().accept(MediaType.TEXT_PLAIN).delete();
      if (DEBUG)
        logger.log(BasicLevel.DEBUG, "RestDistribution.closeProducer: " + uriCloseProducer + " -> " + response.getStatusInfo());
    } catch (Exception exc) {
      if (DEBUG)
        logger.log(BasicLevel.WARN, "RestDistribution.closeProducer, cannot close " + uriCloseProducer, exc);
      else
        logger.log(BasicLevel.WARN, "RestDistribution.closeProducer, cannot close " + uriCloseProducer + ": " + exc.getMessage());
    } finally {
      if (response != null) response.close();
      uriCloseProducer = null;
      // A new client is created in createProducer
      if (client != null) client.close();
      client = null;
    }
  }

  /**
   * Closes the Rest/JMS producer and all associated resources.
   * Called by generic behavior during the destination finalization.
   */
  @Override
  public void close() {
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, "RestDistribution.close()");
    closeProducer();
  }
}
