package batch.controller;

import java.io.IOException;
import java.net.InetAddress;
import java.net.MalformedURLException;
import java.net.UnknownHostException;
import java.rmi.Naming;
import java.rmi.Remote;
import java.rmi.RemoteException;
import java.rmi.registry.LocateRegistry;
import java.rmi.registry.Registry;
import java.rmi.server.UnicastRemoteObject;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

import javax.rmi.ssl.SslRMIClientSocketFactory;
import javax.rmi.ssl.SslRMIServerSocketFactory;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.util.Unbox;

import batch.status.Job;
import batch.status.JobDetail;
import batch.status.JobDetailStatus;
import batch.status.JobMaster;
import batch.status.JobMasterInfo;
import batch.status.JobState;
import batch.status.JobStatus;
import common.db.jdbc.Jdbc;
import core.config.Env;
import core.config.Factory;
import core.exception.PhysicalException;
import core.exception.ThreadExceptionHandler;
import core.exception.ThrowableUtil;
import core.util.BooleanUtil;
import core.util.DateUtil;

/**
 * バッチジョブ管理実装
 *
 * @author Tadashi Nakayama
 * @version 1.0.0
 */
public class JobManagerImpl extends UnicastRemoteObject implements JobManager, Runnable {
	/** serialVersionUID */
	private static final long serialVersionUID = 4825560113231843758L;

	/** ジョブ拡張子 */
	private static final String ENV_BATCH_SUFFIX = "Batch.Suffix";
	/** バッチ起動最大数 */
	private static final String ENV_BATCH_MAXPROCS = "Batch.MaxProcs";
	/** バッチ起動間隔 */
	private static final String ENV_BATCH_INTERVAL = "Batch.Interval";
	/** 中止コマンド */
	private static final String ENV_BATCH_KILL = "Batch.KillCommand";

	/** ジョブ管理マスタ */
	private final Map<String, JobMaster> jobMasterMap = new HashMap<>();
	/** デフォルト最大実行ジョブ数 */
	private final int maxProc;

	/**
	 * コンストラクタ
	 *
	 * @throws RemoteException リモート例外
	 */
	public JobManagerImpl() throws RemoteException {
		// 最大実行ジョブ数取得
		this.maxProc = Env.getEnv(ENV_BATCH_MAXPROCS, 20);
		LogManager.getLogger().info("MaxProc={}", Unbox.box(this.maxProc));
	}

	/**
	 * @see batch.controller.JobManager#cancel(long)
	 */
	@Override
	public void cancel(final long jobseq) throws RemoteException {
		final JobStatus js = Factory.create(JobStatus.class);

		try (Connection conn = JobUtil.getConnection()) {
			conn.clearWarnings();
			conn.setReadOnly(false);
			conn.setAutoCommit(false);

			// ジョブ管理排他読込
			final Job job = js.getJobWithLock(conn, jobseq);
			if (job == null) {
				return;
			}

			final JobState sts = JobState.cancel(job.getJobSts());
			if (sts != null) {
				cancelJob(job.getJobId(), jobseq, conn);

				// ジョブ管理更新
				js.updateJobStatus(conn, jobseq, job.getHostId(), "", sts);

				conn.commit();
			}

		} catch (final SQLException ex) {
			ThrowableUtil.error(ex);
			throw new RemoteException(ex.getMessage(), ex);
		}
	}

	/**
	 * ジョブキャンセル
	 *
	 * @param jobid ジョブID
	 * @param jobseq ジョブ連番
	 * @param conn コネクション
	 */
	private void cancelJob(final String jobid, final long jobseq, final Connection conn) {
		// 終了タイプ取得
		final String type = Optional.ofNullable(this.jobMasterMap.get(jobid)).
				map(JobMaster::getCancelType).orElse("");

		LogManager.getLogger().debug("type={}", type);

		if (!BooleanUtil.toBool(type)) {
			return;
		}

		// ジョブ詳細管理排他読込
		final JobDetailStatus jds = Factory.create(JobDetailStatus.class);
		final List<JobDetail> l = jds.selectJobDetails(conn, jobseq);

		if (l != null && !l.isEmpty() && JobState.ID_B_EDIT.value() == l.get(0).getBatSts()) {
			// 中止コマンド作成
			final String cmd = Env.getEnv(ENV_BATCH_KILL);

			LogManager.getLogger().debug("job detail status={} command={}",
					Unbox.box(l.get(0).getBatSts()), cmd);

			if (!Objects.toString(cmd, "").trim().isEmpty()) {
				// ジョブ中止処理
				try {
					new ProcessBuilder(cmd, String.valueOf(l.get(0).getPid())).start();
				} catch (final IOException ex) {
					LogManager.getLogger().warn(ex.getMessage(), ex);
				} finally {
					LogManager.getLogger().debug("executed={}", cmd);
				}
			}
		}
	}

	/**
	 * プロセス起動管理処理
	 *
	 */
	@Override
	public void run() {
		LogManager.getLogger().info("JobManager Thread Start.");

		// ポーリング処理
		try {
			final InetAddress localhost = InetAddress.getLocalHost();
			launch(localhost.getHostName());
		} catch (final UnknownHostException ex) {
			LogManager.getLogger().error(ex.getMessage(), ex);
			throw new PhysicalException(ex);
		}

		LogManager.getLogger().info("JobManager Thread End.");
	}

	/**
	 * ジョブ管理マスタ情報取得
	 */
	private void getJobMaster() {
		try (Jdbc conn = JobUtil.getConnection()) {
			// ジョブ管理マスタ取得
			final JobMasterInfo ji = Factory.create(JobMasterInfo.class);
			final List<JobMaster> l = ji.getJobMasterInfoAll(conn);
			if (l != null) {
				for (final JobMaster obj : l) {
					this.jobMasterMap.put(obj.getJobId(), obj);
				}
			}
		}
	}

	/**
	 * 起動処理
	 *
	 * @param host ホスト名
	 * @return 起動数
	 */
	private int launch(final String host) {

		int launchedProc = 0;

		try (Connection conn = JobUtil.getConnection()) {
			// ジョブステータス取得
			final JobStatus js = Factory.create(JobStatus.class);
			final List<Job> active = js.getJobListByStatus(conn, host, JobState.running());
			// 実行プロセス数チェック
			if (this.maxProc <= active.size()) {
				return launchedProc;
			}

			// 処理依頼中のプロセスを登録日時順に取得
			final List<Job> queue = js.getJobListByStatus(conn, null, JobState.requested());

			// 最大実行ジョブ数になるか、処理依頼中がなくなるまでプロセス起動
			LogManager.getLogger().debug("RequestCount={}", Unbox.box(queue.size()));

			// 現在時刻取得
			final String now = DateUtil.format(new Date(), "HHmm");

			for (final Job ap : queue) {
				final String jid = ap.getJobId().trim();

				// 中止依頼
				// 起動可能時刻チェック
				if (!checkLaunchTime(jid, now)) {
					continue;
				}

				// ジョブの実行多重度チェック
				final Integer count = Optional.ofNullable(this.jobMasterMap.get(jid)).
						map(JobMaster::getMultiplicity).orElse(Integer.valueOf(1));
				if (count.intValue() <= getActiveProcCount(active, jid)) {
					continue;
				}

				// バッチ起動シェルの実行
				if (execBatchShell(ap.getJobSeq(), jid, host, conn)) {
					launchedProc++;
					active.add(ap);
				}

				conn.commit();
			}

		} catch (final SQLException ex) {
			ThrowableUtil.error(ex);
		}

		return launchedProc;
	}

	/**
	 * ジョブ起動時間チェック
	 *
	 * @param jid ジョブID
	 * @param now 現在時刻
	 * @return 起動可否
	 */
	private boolean checkLaunchTime(final String jid, final String now) {
		final JobMaster rec = this.jobMasterMap.get(jid);
		return rec == null
				|| JobUtil.checkInTime(now, rec.getRunnableBegin(), rec.getRunnableEnd());
	}

	/**
	 * 処理中プロセス数取得
	 *
	 * @param list リスト
	 * @param jid ジョブID
	 * @return 処理依頼中プロセスVector
	 */
	private int getActiveProcCount(final List<Job> list, final String jid) {
		int ret = 0;
		if (list != null) {
			for (final Job vals : list) {
				if (jid.equals(vals.getJobId())) {
					ret++;
				}
			}
		}
		return ret;
	}

	/**
	 * バッチ起動シェル実行
	 *
	 * @param jobseq ジョブ連番
	 * @param jid ジョブID
	 * @param host ホスト名
	 * @param conn コネクション
	 * @return 実行結果
	 */
	private boolean execBatchShell(final long jobseq, final String jid,
			final String host, final Connection conn) {
		// レコードにロックをかける
		final JobStatus js = Factory.create(JobStatus.class);
		final Job job = js.getJobWithLock(conn, jobseq);
		if (job == null || !JobState.isRequest(job.getJobSts())) {
			return false;
		}

		JobState sts = JobState.ID_B_SH_START;
		try {
			new ProcessBuilder(newCommands(jobseq, jid, job)).start();
		} catch (final IOException ex) {
			LogManager.getLogger().error(ex.getMessage(), ex);
			sts = JobState.ID_B_INVALID;
		}

		// バッチ管理更新
		js.updateJobStatus(conn, jobseq, host, "", sts);

		return JobState.ID_B_SH_START.equals(sts);
	}

	/**
	 * コマンドリスト作成
	 *
	 * @param jobseq ジョブ連番
	 * @param jid ジョブID
	 * @param job ジョブ管理
	 * @return コマンドリスト
	 */
	private List<String> newCommands(final long jobseq, final String jid, final Job job) {
		// バッチ起動シェルコマンド作成
		final String suffix = Env.getEnv(ENV_BATCH_SUFFIX);

		LogManager.getLogger().debug(() -> jid + suffix + " " + ID_B_ONLINE_OPTION + " " + jobseq);

		// バッチ起動処理
		final List<String> cmd = new ArrayList<>();
		cmd.add(jid + suffix);
		cmd.add(ID_B_ONLINE_OPTION);
		cmd.add(String.valueOf(jobseq));
		cmd.addAll(Arrays.asList(job.getJobParam()));
		return cmd;
	}

	//	/**
	//	 * クリーン指定
	//	 * @param args 引数
	//	 */
	//	private static void clean(final String... args) {
	//		for (final String arg : args) {
	//			if ("-clean".equals(arg)) {
	//				// TODO DB整合処理
	//				return;
	//			}
	//		}
	//	}

	/**
	 * JobManager取得
	 */
	private void start() {
		// スレッド起動
		final ScheduledExecutorService ses =
						Executors.newSingleThreadScheduledExecutor(new JobThreadFactory());
		ses.scheduleWithFixedDelay(this, TimeUnit.SECONDS.toMillis(30),
						Env.getEnv(ENV_BATCH_INTERVAL, 10000), TimeUnit.MILLISECONDS);
	}

	/**
	 * メイン処理
	 *
	 * @param args 起動パラメタ
	 */
	public static void main(final String... args) {
		// セキュリティマネージャ設定
		if (System.getSecurityManager() == null) {
			System.setSecurityManager(new SecurityManager());
		}

		// clean(args);

		// バインディング
		try {
			// JobManager取得
			final JobManagerImpl jm = new JobManagerImpl();
			jm.getJobMaster();
			jm.start();

			final int ssl = JobRequestor.getSslPort();
			if (0 < ssl) {
				final Remote obj = UnicastRemoteObject.exportObject(jm, ssl,
								new SslRMIClientSocketFactory(),
								new SslRMIServerSocketFactory());

				final Registry registry = LocateRegistry.createRegistry(ssl,
								new SslRMIClientSocketFactory(),
								new SslRMIServerSocketFactory());
				registry.rebind("//localhost" + BATCH_BIND_NAME, obj);
			} else {
				Naming.rebind(JobRequestor.toBatchUri("//localhost", BATCH_BIND_NAME), jm);
			}

			LogManager.getLogger().debug("End Bind");

		} catch (final RemoteException | MalformedURLException ex) {
			ThrowableUtil.error(ex);
			Runtime.getRuntime().exit(1);
		}
	}

	/**
	 * JobThreadFactory
	 * @author Tadashi Nakayama
	 */
	private static final class JobThreadFactory implements ThreadFactory {
		/**
		 * コンストラクタ
		 */
		JobThreadFactory() {
			super();
		}

		/**
		 * @see java.util.concurrent.ThreadFactory#newThread(java.lang.Runnable)
		 */
		@Override
		public Thread newThread(final Runnable r) {
			final Thread th = new Thread(r, "JobManagerImpl");
			th.setUncaughtExceptionHandler(new ThreadExceptionHandler());
			return th;
		}
	}
}
