// Copyright 2011 Google Inc. All Rights Reserved.
//
// Licensed 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 adaptorlib;

import com.sun.net.httpserver.HttpExchange;
import com.sun.net.httpserver.HttpServer;
import com.sun.net.httpserver.HttpsConfigurator;
import com.sun.net.httpserver.HttpsParameters;
import com.sun.net.httpserver.HttpsServer;

import java.io.Closeable;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.TreeMap;
import java.util.concurrent.Executors;
import java.util.logging.Level;
import java.util.logging.LogManager;
import java.util.logging.Logger;

import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLParameters;

/** This class handles the communications with GSA. */
public class GsaCommunicationHandler implements DocIdEncoder, DocIdDecoder {
  private static final Logger log
      = Logger.getLogger(GsaCommunicationHandler.class.getName());

  private final Adaptor adaptor;
  private final Config config;
  private final GsaFeedFileSender fileSender;
  private final GsaFeedFileMaker fileMaker;
  private final Journal journal = new Journal();
  private final Adaptor.DocIdPusher pusher = new InnerDocIdPusher();
  private final Adaptor.GetDocIdsErrorHandler defaultErrorHandler
      = new DefaultGetDocIdsErrorHandler();
  private Scheduler pushScheduler;
  private int pushingDocIds;
  private HttpServer server;
  private CircularLogRpcMethod circularLogRpcMethod;
  private Thread shutdownHook;

  public GsaCommunicationHandler(Adaptor adaptor, Config config) {
    // TODO(ejona): allow the adaptor to choose whether it wants this feature
    this.adaptor = new AutoUnzipAdaptor(adaptor);
    this.adaptor.setDocIdPusher(pusher);

    this.config = config;
    this.fileSender = new GsaFeedFileSender(config.getGsaCharacterEncoding(),
                                            config.isServerSecure());
    this.fileMaker = new GsaFeedFileMaker(this);
  }

  /** Starts listening for communications from GSA. */
  public synchronized void beginListeningForContentRequests()
      throws IOException {
    if (server != null) {
      throw new IllegalStateException("Already listening");
    }

    int port = config.getServerPort();
    boolean secure = config.isServerSecure();
    InetSocketAddress addr = new InetSocketAddress(port);
    if (!secure) {
      server = HttpServer.create(addr, 0);
    } else {
      server = HttpsServer.create(addr, 0);
      try {
        HttpsConfigurator httpsConf
            = new HttpsConfigurator(SSLContext.getDefault()) {
              public void configure(HttpsParameters params) {
                SSLParameters sslParams
                    = getSSLContext().getDefaultSSLParameters();
                // Allow verifying the GSA and other trusted computers.
                sslParams.setWantClientAuth(true);
                params.setSSLParameters(sslParams);
              }
            };
        ((HttpsServer) server).setHttpsConfigurator(httpsConf);
      } catch (java.security.NoSuchAlgorithmException ex) {
        throw new RuntimeException(ex);
      }
    }
    // If the port is zero, then the OS chose a port for us. This is mainly
    // useful during testing.
    port = server.getAddress().getPort();
    config.setValue("server.port", "" + port);
    server.createContext("/dashboard", new DashboardHandler(config, journal));
    server.createContext("/rpc", createRpcHandler());

    SessionManager<HttpExchange> sessionManager = null;
    AuthnHandler authnHandler = null;
    if (secure) {
       sessionManager = new SessionManager<HttpExchange>(
          new SessionManager.HttpExchangeCookieAccess(),
          30 * 60 * 1000 /* session lifetime: 30 minutes */,
          5 * 60 * 1000 /* max cleanup frequency: 5 minutes */);
      server.createContext("/samlassertionconsumer",
          new SamlAssertionConsumerHandler(config.getServerHostname(),
            config.getGsaCharacterEncoding(), sessionManager));
      authnHandler = new AuthnHandler(config.getServerHostname(),
          config.getGsaCharacterEncoding(), sessionManager,
          config.getGsaHostname(), config.getServerKeyAlias(),
          config.getServerPort());
    }
    server.createContext(config.getServerBaseUri().getPath()
        + config.getServerDocIdPath(),
        new DocumentHandler(config.getServerHostname(),
                            config.getGsaCharacterEncoding(), this,
                            getJournal(), adaptor,
                            config.getServerAddResolvedGsaHostnameToGsaIps(),
                            config.getGsaHostname(), config.getServerGsaIps(),
                            authnHandler, sessionManager));
    server.setExecutor(Executors.newCachedThreadPool());
    server.start();
    log.info("GSA host name: " + config.getGsaHostname());
    log.info("server is listening on port #" + port);
    shutdownHook = new Thread(new ShutdownHook(), "gsacomm-shutdown");
    Runtime.getRuntime().addShutdownHook(shutdownHook);
  }

  /**
   * Stop the current services, allowing up to {@code maxDelay} seconds for
   * things to shutdown.
   */
  public synchronized void stop(int maxDelay) {
    if (circularLogRpcMethod != null) {
      circularLogRpcMethod.close();
      circularLogRpcMethod = null;
    }
    if (shutdownHook != null) {
      try {
        Runtime.getRuntime().removeShutdownHook(shutdownHook);
      } catch (IllegalStateException ex) {
        // Already executing hook.
      }
      shutdownHook = null;
    }
    if (pushScheduler != null) {
      pushScheduler.cancel();
      pushScheduler = null;
    }
    if (server != null) {
      server.stop(maxDelay);
      server = null;
    }
  }

  private synchronized RpcHandler createRpcHandler() {
    RpcHandler rpcHandler = new RpcHandler(config.getServerHostname(),
        config.getGsaCharacterEncoding(), this);
    rpcHandler.registerRpcMethod("startFeedPush", new StartFeedPushRpcMethod());
    circularLogRpcMethod = new CircularLogRpcMethod();
    rpcHandler.registerRpcMethod("getLog", circularLogRpcMethod);
    rpcHandler.registerRpcMethod("getConfig", new ConfigRpcMethod(config));
    rpcHandler.registerRpcMethod("getStats", new StatRpcMethod(journal));
    return rpcHandler;
  }

  /**
   * Schedule {@link Adaptor#getDocIds} to be called when defined by the {@code
   * schedule}. Equivalent to {@code beginPushingDocIds(schedule, null)}.
   *
   * @see #beginPushingDocIds(Iterator, Adaptor.GetDocIdsErrorHandler)
   */
  public void beginPushingDocIds(Iterator<Date> schedule) {
    beginPushingDocIds(schedule, null);
  }

  /**
   * Schedule {@link Adaptor#getDocIds} to be called when defined by the {@code
   * schedule}. If {@code handler} is {@code null}, then a default error handler
   * will be used.
   */
  public void beginPushingDocIds(Iterator<Date> schedule,
                                 Adaptor.GetDocIdsErrorHandler handler) {
    getPushScheduler().schedule(new PushTask(handler), schedule);
  }

  /**
   * Ensure there is a push running right now. This schedules a new push if one
   * is not already running. Returns {@code true} if it starts a new push, and
   * false otherwise.
   */
  synchronized boolean checkAndBeginPushDocIdsImmediately(
      Adaptor.GetDocIdsErrorHandler handler) {
    if (pushingDocIds > 0) {
      return false;
    }
    beginPushingDocIdsImmediately(handler);
    return true;
  }

  /**
   * Schedule a push for immediately. If there is a push already running this
   * push will be started after it.
   */
  private void beginPushingDocIdsImmediately(
      Adaptor.GetDocIdsErrorHandler handler) {
    List<Date> schedule = Collections.singletonList(new Date());
    getPushScheduler().schedule(new PushTask(handler), schedule.iterator());
  }

  private DocInfo pushSizedBatchOfDocInfos(List<DocInfo> docInfos,
                                           Adaptor.PushErrorHandler handler)
      throws InterruptedException {
    String feedSourceName = config.getFeedName();
    String xmlFeedFile = fileMaker.makeMetadataAndUrlXml(
        feedSourceName, docInfos);
    boolean keepGoing = true;
    boolean success = false;
    for (int ntries = 1; keepGoing; ntries++) {
      try {
        log.info("Sending feed to GSA host name: " + config.getGsaHostname());
        fileSender.sendMetadataAndUrl(config.getGsaHostname(), feedSourceName,
                                      xmlFeedFile);
        keepGoing = false;  // Sent.
        success = true;
      } catch (GsaFeedFileSender.FailedToConnect ftc) {
        log.log(Level.WARNING, "Unable to connect to the GSA", ftc);
        keepGoing = handler.handleFailedToConnect(
            (Exception) ftc.getCause(), ntries);
      } catch (GsaFeedFileSender.FailedWriting fw) {
        log.log(Level.WARNING, "Unable to write request to the GSA", fw);
        keepGoing = handler.handleFailedWriting(
            (Exception) fw.getCause(), ntries);
      } catch (GsaFeedFileSender.FailedReadingReply fr) {
        log.log(Level.WARNING, "Unable to read reply from GSA", fr);
        keepGoing = handler.handleFailedReadingReply(
            (Exception) fr.getCause(), ntries);
      }
      if (keepGoing) {
        log.log(Level.INFO, "Trying again... Number of attemps: {0}", ntries);
      }
    }
    return success ? null : docInfos.get(0);
  }

  /**
   * Makes and sends metadata-and-url feed files to GSA. This method blocks
   * until all DocIds are sent or retrying failed. Equivalent to {@code
   * pushDocIds(null)}.
   */
  public void pushDocIds() throws InterruptedException {
    pushDocIds(null);
  }

  /**
   * Makes and sends metadata-and-url feed files to GSA. This method blocks
   * until all DocIds are sent or retrying failed. If {@code handler} is {@code
   * null}, then a default error handler is used.
   */
  public void pushDocIds(Adaptor.GetDocIdsErrorHandler handler)
      throws InterruptedException {
    synchronized (this) {
      pushingDocIds++;
    }
    if (handler == null) {
      handler = defaultErrorHandler;
    }
    log.info("Getting list of DocIds");
    for (int ntries = 1;; ntries++) {
      boolean keepGoing = true;
      try {
        adaptor.getDocIds(pusher);
        break; // Success
      } catch (Exception ex) {
        log.log(Level.WARNING, "Unable to retrieve DocIds from adaptor", ex);
        keepGoing = handler.handleFailedToGetDocIds(ex, ntries);
      }
      if (keepGoing) {
        log.log(Level.INFO, "Trying again... Number of attemps: {0}", ntries);
      } else {
        return; // Bail
      }
    }
    synchronized (this) {
      pushingDocIds--;
    }
  }

  /**
   * Makes and sends metadata-and-url feed files to GSA. Generally, you should
   * use {@link #pushDocIds()} instead of this method. However, if you want to
   * push just a few DocIds to the GSA manually, this is the method to use.
   * This method blocks until all DocIds are sent or retrying failed.
   */
  private DocInfo pushDocInfos(Iterator<DocInfo> docInfos,
                               Adaptor.PushErrorHandler handler)
      throws InterruptedException {
    log.log(Level.INFO, "Pushing DocIds");
    final int max = config.getFeedMaxUrls();
    while (docInfos.hasNext()) {
      List<DocInfo> batch = new ArrayList<DocInfo>();
      for (int j = 0; j < max; j++) {
        if (!docInfos.hasNext()) {
          break;
        }
        batch.add(docInfos.next());
      }
      log.log(Level.INFO, "Pushing group of {0} DocIds", batch.size());
      DocInfo failedId = pushSizedBatchOfDocInfos(batch, handler);
      if (failedId != null) {
        log.info("Failed to push all ids. Failed on docId: " + failedId);
        return failedId;
      }
      journal.recordDocIdPush(batch);
    }
    log.info("Pushed DocIds");
    return null;
  }

  public URI encodeDocId(DocId docId) {
    if (config.isDocIdUrl()) {
      return URI.create(docId.getUniqueId());
    } else {
      URI base = config.getServerBaseUri(docId);
      URI resource;
      String uniqueId = docId.getUniqueId();
      // Add two dots to any sequence of only dots. This is to allow "/../" and
      // "/./" within DocIds.
      uniqueId = uniqueId.replaceAll("(^|/)(\\.+)(?=$|/)", "$1$2..");
      try {
        resource = new URI(null, null, base.getPath()
                           + config.getServerDocIdPath() + uniqueId, null);
      } catch (URISyntaxException ex) {
        throw new IllegalStateException(ex);
      }
      return base.resolve(resource);
    }
  }

  /** Given a URI that was used in feed file, convert back to doc id. */
  public DocId decodeDocId(URI uri) {
    if (config.isDocIdUrl()) {
      return new DocId(uri.toString());
    } else {
      String basePath = config.getServerBaseUri().getPath();
      String id = uri.getPath().substring(basePath.length()
          + config.getServerDocIdPath().length());
      // Remove two dots from any sequence of only dots. This is to remove the
      // addition we did in {@link #encodeDocId}.
      id = id.replaceAll("(^|/)(\\.+)\\.\\.(?=$|/)", "$1$2");
      return new DocId(id);
    }
  }

  URI formNamespacedUri(String namespace) {
    URI uri;
    try {
      uri = new URI(null, null, config.getServerBaseUri().getPath() + namespace,
                    null);
    } catch (URISyntaxException e) {
      throw new IllegalStateException(e);
    }
    return config.getServerBaseUri().resolve(uri);
  }

  Journal getJournal() {
    return journal;
  }

  private Scheduler getPushScheduler() {
    if (pushScheduler == null) {
      pushScheduler = new Scheduler();
    }
    return pushScheduler;
  }

  private class InnerDocIdPusher extends AbstractDocIdPusher {
    private Adaptor.PushErrorHandler defaultErrorHandler
        = new DefaultPushErrorHandler();

    @Override
    public DocInfo pushDocInfos(Iterable<DocInfo> docInfos,
                                Adaptor.PushErrorHandler handler)
        throws InterruptedException {
      if (handler == null) {
        handler = defaultErrorHandler;
      }
      return GsaCommunicationHandler.this.pushDocInfos(docInfos.iterator(),
                                                       handler);
    }
  }

  private class PushTask extends Scheduler.Task {
    private final Adaptor.GetDocIdsErrorHandler handler;

    public PushTask(Adaptor.GetDocIdsErrorHandler handler) {
      this.handler = handler;
    }

    public void run() {
      try {
        pushDocIds(handler);
      } catch (InterruptedException ex) {
        Thread.currentThread().interrupt();
      }
    }
  }

  private class ShutdownHook implements Runnable {
    @Override
    public void run() {
      // Allow three seconds for things to stop.
      stop(3);
    }
  }

  class StartFeedPushRpcMethod implements RpcHandler.RpcMethod {
    @Override
    public Object run(List request) {
      boolean pushStarted = checkAndBeginPushDocIdsImmediately(null);
      if (!pushStarted) {
        throw new RuntimeException("A push is already in progress");
      }
      return 1;
    }
  }

  static class CircularLogRpcMethod implements RpcHandler.RpcMethod, Closeable {
    private final CircularBufferHandler circularLog
        = new CircularBufferHandler();

    /**
     * Installs a log handler; to uninstall handler, call {@link #close}.
     */
    public CircularLogRpcMethod() {
      LogManager.getLogManager().getLogger("").addHandler(circularLog);
    }

    @Override
    public Object run(List request) {
      return circularLog.writeOut();
    }

    @Override
    public void close() {
      LogManager.getLogManager().getLogger("").removeHandler(circularLog);
    }
  }

  static class ConfigRpcMethod implements RpcHandler.RpcMethod {
    private final Config config;

    public ConfigRpcMethod(Config config) {
      this.config = config;
    }

    public Object run(List request) {
      TreeMap<String, String> configMap = new TreeMap<String, String>();
      for (String key : config.getAllKeys()) {
        configMap.put(key, config.getValue(key));
      }
      return configMap;
    }
  }
}