/*
 * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
 */

package io.airbyte.workers.process;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Sets;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;
import javax.annotation.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Convenience wrapper around a thread-safe BlockingQueue. Keeps track of available ports for Kube
 * Pod Processes.
 *
 * Although this data structure can do without the wrapper class, this class allows easier testing
 * via the {@link #getNumAvailablePorts()} function.
 *
 * The singleton pattern clarifies that only one copy of this class is intended to exist per
 * scheduler deployment.
 */
public class KubePortManagerSingleton {

  private static final Logger LOGGER = LoggerFactory.getLogger(KubePortManagerSingleton.class);

  private static KubePortManagerSingleton instance;

  private static final int MAX_PORTS_PER_WORKER = 4; // A sync has two workers. Each worker requires 2 ports.
  private final BlockingQueue<Integer> workerPorts;

  private KubePortManagerSingleton(final Set<Integer> ports) {
    workerPorts = new LinkedBlockingDeque<>(ports);
  }

  /**
   * Make sure init(ports) is called once prior to repeatedly using getInstance().
   */
  public static synchronized KubePortManagerSingleton getInstance() {
    if (instance == null) {
      throw new RuntimeException("Must initialize with init(ports) before using.");
    }
    return instance;
  }

  /**
   * Sets up the port range; make sure init(ports) is called once prior to repeatedly using
   * getInstance(). Init won't fail (it will perform a no-op) if re-initializd with the same set of
   * ports.
   */
  public static synchronized void init(final Set<Integer> ports) {
    if (instance == null) {
      instance = new KubePortManagerSingleton(ports);
    } else if (Sets.intersection(instance.getAllPorts(), ports).size() != ports.size()) {
      LOGGER.info("Skipping initializing KubePortManagerSingleton since ports specified are the same.");
    } else {
      throw new RuntimeException("Cannot initialize twice with different ports!");
    }
  }

  public Integer take() throws InterruptedException {
    return workerPorts.poll(10, TimeUnit.MINUTES);
  }

  @VisibleForTesting
  public @Nullable Integer takeImmediately() {
    return workerPorts.poll();
  }

  public void offer(final Integer port) {
    if (!workerPorts.contains(port)) {
      workerPorts.add(port);
    }
  }

  protected Set<Integer> getAllPorts() {
    return new HashSet<>(workerPorts);
  }

  public int getNumAvailablePorts() {
    return workerPorts.size();
  }

  public int getSupportedWorkers() {
    return workerPorts.size() / MAX_PORTS_PER_WORKER;
  }

}
