/*
    * Copyright (C) 2019 Google LLC
    * and other copyright owners as documented in the project's IP log.
    *
    * This program and the accompanying materials are made available
    * under the terms of the Eclipse Distribution License v1.0 which
    * accompanies this distribution, is reproduced below, and is
    * available at http://www.eclipse.org/org/documents/edl-v10.php
    *
   * All rights reserved.
   *
   * Redistribution and use in source and binary forms, with or
   * without modification, are permitted provided that the following
   * conditions are met:
   *
   * - Redistributions of source code must retain the above copyright
   *   notice, this list of conditions and the following disclaimer.
   *
   * - Redistributions in binary form must reproduce the above
   *   copyright notice, this list of conditions and the following
   *   disclaimer in the documentation and/or other materials provided
   *   with the distribution.
   *
   * - Neither the name of the Eclipse Foundation, Inc. nor the
   *   names of its contributors may be used to endorse or promote
   *   products derived from this software without specific prior
   *   written permission.
   *
   * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND
   * CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,
   * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
   * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
   * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR
   * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
   * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
   * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
   * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
   * CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
   * STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
   * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF
   * ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
   */
  
  package org.eclipse.jgit.internal.storage.file;
  
  import static java.nio.charset.StandardCharsets.UTF_8;
  
  import java.io.BufferedReader;
  import java.io.File;
  import java.io.FileInputStream;
  import java.io.FileNotFoundException;
  import java.io.FileOutputStream;
  import java.io.IOException;
  import java.io.InputStreamReader;
  import java.nio.file.Files;
  import java.nio.file.StandardCopyOption;
  import java.util.ArrayList;
  import java.util.Comparator;
  import java.util.List;
  import java.util.Map;
  import java.util.Optional;
  import java.util.function.Supplier;
  import java.util.stream.Collectors;
  
  import org.eclipse.jgit.annotations.Nullable;
  import org.eclipse.jgit.errors.LockFailedException;
  import org.eclipse.jgit.internal.storage.io.BlockSource;
  import org.eclipse.jgit.internal.storage.reftable.MergedReftable;
  import org.eclipse.jgit.internal.storage.reftable.ReftableCompactor;
  import org.eclipse.jgit.internal.storage.reftable.ReftableConfig;
  import org.eclipse.jgit.internal.storage.reftable.ReftableReader;
  import org.eclipse.jgit.internal.storage.reftable.ReftableWriter;
  import org.eclipse.jgit.lib.Config;
  import org.eclipse.jgit.util.FileUtils;
  
  /**
   * A mutable stack of reftables on local filesystem storage. Not thread-safe.
   * This is an AutoCloseable because this object owns the file handles to the
   * open reftables.
   */
  public class FileReftableStack implements AutoCloseable {
  	private static class StackEntry {
  
  		String name;
  
  		ReftableReader reftableReader;
  	}
  
  	private MergedReftable mergedReftable;
  
  	private List<StackEntry> stack;
  
  	private long lastNextUpdateIndex;
  
  	private final File stackPath;
  
  	private final File reftableDir;
  
  	private final Runnable onChange;
 
 	private final Supplier<Config> configSupplier;
 
 	// Used for stats & testing.
 	static class CompactionStats {
 
 		long tables;
 
 		long bytes;
 
 		int attempted;
 
 		int failed;
 
 		long refCount;
 
 		long logCount;
 
 		CompactionStats() {
 			tables = 0;
 			bytes = 0;
 			attempted = 0;
 			failed = 0;
 			logCount = 0;
 			refCount = 0;
 		}
 	}
 
 	private final CompactionStats stats;
 
 	/**
 	 * Creates a stack corresponding to the list of reftables in the argument
 	 *
 	 * @param stackPath
 	 *            the filename for the stack.
 	 * @param reftableDir
 	 *            the dir holding the tables.
 	 * @param onChange
 	 *            hook to call if we notice a new write
 	 * @param configSupplier
 	 *            Config supplier
 	 * @throws IOException
 	 *             on I/O problems
 	 */
 	public FileReftableStack(File stackPath, File reftableDir,
 			@Nullable Runnable onChange, Supplier<Config> configSupplier)
 			throws IOException {
 		this.stackPath = stackPath;
 		this.reftableDir = reftableDir;
 		this.stack = new ArrayList<>();
 		this.configSupplier = configSupplier;
 		this.onChange = onChange;
 
 		// skip event notification
 		lastNextUpdateIndex = 0;
 		reload();
 
 		stats = new CompactionStats();
 	}
 
 	CompactionStats getStats() {
 		return stats;
 	}
 
 	/** Thrown if the update indices in the stack are not monotonic */
 	public static class ReftableNumbersNotIncreasingException
 			extends RuntimeException {
 		private static final long serialVersionUID = 1L;
 
 		String name;
 
 		long lastMax;
 
 		long min;
 
 		ReftableNumbersNotIncreasingException(String name, long lastMax,
 				long min) {
 			this.name = name;
 			this.lastMax = lastMax;
 			this.min = min;
 		}
 	}
 
 	/**
 	 * Reloads the stack, potentially reusing opened reftableReaders.
 	 *
 	 * @param names
 	 *            holds the names of the tables to load.
 	 * @throws FileNotFoundException
 	 *             load must be retried.
 	 * @throws IOException
 	 *             on other IO errors.
 	 */
 	private void reloadOnce(List<String> names)
 			throws IOException, FileNotFoundException {
 		Map<String, ReftableReader> current = stack.stream()
 				.collect(Collectors.toMap(e -> e.name, e -> e.reftableReader));
 
 		List<ReftableReader> newTables = new ArrayList<>();
 		List<StackEntry> newStack = new ArrayList<>(stack.size() + 1);
 		try {
 			ReftableReader last = null;
 			for (String name : names) {
 				StackEntry entry = new StackEntry();
 				entry.name = name;
 
 				ReftableReader t = null;
 				if (current.containsKey(name)) {
 					t = current.remove(name);
 				} else {
 					File subtable = new File(reftableDir, name);
 					FileInputStream is;
 
 					is = new FileInputStream(subtable);
 
 					t = new ReftableReader(BlockSource.from(is));
 					newTables.add(t);
 				}
 
 				if (last != null) {
 					// TODO: move this to MergedReftable
 					if (last.maxUpdateIndex() >= t.minUpdateIndex()) {
 						throw new ReftableNumbersNotIncreasingException(name,
 								last.maxUpdateIndex(), t.minUpdateIndex());
 					}
 				}
 				last = t;
 
 				entry.reftableReader = t;
 				newStack.add(entry);
 			}
 			// survived without exceptions: swap in new stack, and close
 			// dangling tables.
 			stack = newStack;
 			newTables.clear();
 
 			current.values().forEach(r -> {
 				try {
 					r.close();
 				} catch (IOException e) {
 					throw new AssertionError(e);
 				}
 			});
 		} finally {
 			newTables.forEach(t -> {
 				try {
 					t.close();
 				} catch (IOException ioe) {
 					// reader close should not generate errors.
 					throw new AssertionError(ioe);
 				}
 			});
 		}
 	}
 
 	void reload() throws IOException {
 		// Try for 2.5 seconds.
 		long deadline = System.currentTimeMillis() + 2500;
 		// A successful reftable transaction is 2 atomic file writes
 		// (open, write, close, rename), which a fast Linux system should be
 		// able to do in about ~200us. So 1 ms should be ample time.
 		long min = 1;
 		long max = 1000;
 		long delay = 0;
 		boolean success = false;
 		while (System.currentTimeMillis() < deadline) {
 			List<String> names = readTableNames();
 			try {
 				reloadOnce(names);
 				success = true;
 				break;
 			} catch (FileNotFoundException e) {
 				List<String> changed = readTableNames();
 				if (changed.equals(names)) {
 					throw e;
 				}
 			}
 
 			delay = FileUtils.delay(delay, min, max);
 			try {
 				Thread.sleep(delay);
 			} catch (InterruptedException e) {
 				Thread.currentThread().interrupt();
 				throw new RuntimeException(e);
 			}
 		}
 
 		if (!success) {
 			// TODO: should reexamine the 'refs' file to see if it was the same
 			// if it didn't change, then we must have corruption. If it did,
 			// retry.
 			throw new LockFailedException(stackPath);
 		}
 
 		mergedReftable = new MergedReftable(stack.stream()
 				.map(x -> x.reftableReader).collect(Collectors.toList()));
 		long curr = nextUpdateIndex();
 		if (lastNextUpdateIndex > 0 && lastNextUpdateIndex != curr
 				&& onChange != null) {
 			onChange.run();
 		}
 		lastNextUpdateIndex = curr;
 	}
 
 	/**
 	 * @return the merged reftable
 	 */
 	public MergedReftable getMergedReftable() {
 		return mergedReftable;
 	}
 
 	/**
 	 * Writer is a callable that writes data to a reftable under construction.
 	 * It should set the min/max update index, and then write refs and/or logs.
 	 * It should not call finish() on the writer.
 	 */
 	public interface Writer {
 		/**
 		 * Write data to reftable
 		 *
 		 * @param w
 		 *            writer to use
 		 * @throws IOException
 		 */
 		void call(ReftableWriter w) throws IOException;
 	}
 
 	private List<String> readTableNames() throws IOException {
 		List<String> names = new ArrayList<>(stack.size() + 1);
 
 		try (BufferedReader br = new BufferedReader(
 				new InputStreamReader(new FileInputStream(stackPath), UTF_8))) {
 			String line;
 			while ((line = br.readLine()) != null) {
 				if (!line.isEmpty()) {
 					names.add(line);
 				}
 			}
 		} catch (FileNotFoundException e) {
 			// file isn't there: empty repository.
 		}
 		return names;
 	}
 
 	/**
 	 * @return true if the on-disk file corresponds to the in-memory data.
 	 * @throws IOException
 	 *             on IO problem
 	 */
 	boolean isUpToDate() throws IOException {
 		// We could use FileSnapshot to avoid reading the file, but the file is
 		// small so it's probably a minor optimization.
 		try {
 			List<String> names = readTableNames();
 			if (names.size() != stack.size()) {
 				return false;
 			}
 			for (int i = 0; i < names.size(); i++) {
 				if (!names.get(i).equals(stack.get(i).name)) {
 					return false;
 				}
 			}
 		} catch (FileNotFoundException e) {
 			return stack.isEmpty();
 		}
 		return true;
 	}
 
 	/**
 	 * {@inheritDoc}
 	 */
 	@Override
 	public void close() {
 		for (StackEntry entry : stack) {
 			try {
 				entry.reftableReader.close();
 			} catch (Exception e) {
 				// we are reading; this should never fail.
 				throw new AssertionError(e);
 			}
 		}
 	}
 
 	private long nextUpdateIndex() throws IOException {
 		return stack.size() > 0
 				? stack.get(stack.size() - 1).reftableReader.maxUpdateIndex()
 						+ 1
 				: 1;
 	}
 
 	private String filename(long low, long high) {
 		return String.format("%012x-%012x", //$NON-NLS-1$
 				Long.valueOf(low), Long.valueOf(high));
 	}
 
 	/**
 	 * Tries to add a new reftable to the stack. Returns true if it succeeded,
 	 * or false if there was a lock failure, due to races with other processes.
 	 * This is package private so FileReftableDatabase can call into here.
 	 *
 	 * @param w
 	 *            writer to write data to a reftable under construction
 	 * @return true if the transaction.
 	 * @throws IOException
 	 *             on I/O problems
 	 */
 	@SuppressWarnings("nls")
 	public boolean addReftable(Writer w) throws IOException {
 		LockFile lock = new LockFile(stackPath);
 		try {
 			if (!lock.lockForAppend()) {
 				return false;
 			}
 			if (!isUpToDate()) {
 				return false;
 			}
 
 			String fn = filename(nextUpdateIndex(), nextUpdateIndex());
 
 			File tmpTable = File.createTempFile(fn + "_", ".ref",
 					stackPath.getParentFile());
 
 			ReftableWriter.Stats s;
 			try (FileOutputStream fos = new FileOutputStream(tmpTable)) {
 				ReftableWriter rw = new ReftableWriter(reftableConfig(), fos);
 				w.call(rw);
 				rw.finish();
 				s = rw.getStats();
 			}
 
 			if (s.minUpdateIndex() < nextUpdateIndex()) {
 				return false;
 			}
 
 			// The spec says to name log-only files with .log, which is somewhat
 			// pointless given compaction, but we do so anyway.
 			fn += s.refCount() > 0 ? ".ref" : ".log";
 			File dest = new File(reftableDir, fn);
 
 			FileUtils.rename(tmpTable, dest, StandardCopyOption.ATOMIC_MOVE);
 			lock.write((fn + "\n").getBytes(UTF_8));
 			if (!lock.commit()) {
 				FileUtils.delete(dest);
 				return false;
 			}
 
 			reload();
 
 			autoCompact();
 		} finally {
 			lock.unlock();
 		}
 		return true;
 	}
 
 	private ReftableConfig reftableConfig() {
 		return new ReftableConfig(configSupplier.get());
 	}
 
 	/**
 	 * Write the reftable for the given range into a temp file.
 	 *
 	 * @param first
 	 *            index of first stack entry to be written
 	 * @param last
 	 *            index of last stack entry to be written
 	 * @return the file holding the replacement table.
 	 * @throws IOException
 	 *             on I/O problem
 	 */
 	private File compactLocked(int first, int last) throws IOException {
 		String fn = filename(first, last);
 
 		File tmpTable = File.createTempFile(fn + "_", ".ref", //$NON-NLS-1$//$NON-NLS-2$
 				stackPath.getParentFile());
 		try (FileOutputStream fos = new FileOutputStream(tmpTable)) {
 			ReftableCompactor c = new ReftableCompactor(fos)
 					.setConfig(reftableConfig())
 					.setMinUpdateIndex(
 							stack.get(first).reftableReader.minUpdateIndex())
 					.setMaxUpdateIndex(
 							stack.get(last).reftableReader.maxUpdateIndex())
 					.setIncludeDeletes(first > 0);
 
 			List<ReftableReader> compactMe = new ArrayList<>();
 			long totalBytes = 0;
 			for (int i = first; i <= last; i++) {
 				compactMe.add(stack.get(i).reftableReader);
 				totalBytes += stack.get(i).reftableReader.size();
 			}
 			c.addAll(compactMe);
 
 			c.compact();
 
 			// Even though the compaction did not definitely succeed, we keep
 			// tally here as we've expended the effort.
 			stats.bytes += totalBytes;
 			stats.tables += first - last + 1;
 			stats.attempted++;
 			stats.refCount += c.getStats().refCount();
 			stats.logCount += c.getStats().logCount();
 		}
 
 		return tmpTable;
 	}
 
 	/**
 	 * Compacts a range of the stack, following the file locking protocol
 	 * documented in the spec.
 	 *
 	 * @param first
 	 *            index of first stack entry to be considered in compaction
 	 * @param last
 	 *            index of last stack entry to be considered in compaction
 	 * @return true if a compaction was successfully applied.
 	 * @throws IOException
 	 *             on I/O problem
 	 */
 	boolean compactRange(int first, int last) throws IOException {
 		if (first >= last) {
 			return true;
 		}
 		LockFile lock = new LockFile(stackPath);
 
 		File tmpTable = null;
 		List<LockFile> subtableLocks = new ArrayList<>();
 
 		try {
 			if (!lock.lock()) {
 				return false;
 			}
 			if (!isUpToDate()) {
 				return false;
 			}
 
 			List<File> deleteOnSuccess = new ArrayList<>();
 			for (int i = first; i <= last; i++) {
 				File f = new File(reftableDir, stack.get(i).name);
 				LockFile lf = new LockFile(f);
 				if (!lf.lock()) {
 					return false;
 				}
 				subtableLocks.add(lf);
 				deleteOnSuccess.add(f);
 			}
 
 			lock.unlock();
 			lock = null;
 
 			tmpTable = compactLocked(first, last);
 
 			lock = new LockFile(stackPath);
 			if (!lock.lock()) {
 				return false;
 			}
 			if (!isUpToDate()) {
 				return false;
 			}
 
 			String fn = filename(
 					stack.get(first).reftableReader.minUpdateIndex(),
 					stack.get(last).reftableReader.maxUpdateIndex());
 
 			// The spec suggests to use .log for log-only tables, and collect
 			// all log entries in a single file at the bottom of the stack. That would
 			// require supporting overlapping ranges for the different tables. For the
 			// sake of simplicity, we simply ignore this and always produce a log +
 			// ref combined table.
 			fn += ".ref"; //$NON-NLS-1$
 			File dest = new File(reftableDir, fn);
 
 			FileUtils.rename(tmpTable, dest, StandardCopyOption.ATOMIC_MOVE);
 			tmpTable = null;
 
 			StringBuilder sb = new StringBuilder();
 
 			for (int i = 0; i < first; i++) {
 				sb.append(stack.get(i).name + "\n"); //$NON-NLS-1$
 			}
 			sb.append(fn + "\n"); //$NON-NLS-1$
 			for (int i = last + 1; i < stack.size(); i++) {
 				sb.append(stack.get(i).name + "\n"); //$NON-NLS-1$
 			}
 
 			lock.write(sb.toString().getBytes(UTF_8));
 			if (!lock.commit()) {
 				dest.delete();
 				return false;
 			}
 
 			for (File f : deleteOnSuccess) {
 				Files.delete(f.toPath());
 			}
 
 			reload();
 			return true;
 		} finally {
 			if (tmpTable != null) {
 				tmpTable.delete();
 			}
 			for (LockFile lf : subtableLocks) {
 				lf.unlock();
 			}
 			if (lock != null) {
 				lock.unlock();
 			}
 		}
 	}
 
 	/**
 	 * Calculate an approximate log2.
 	 *
 	 * @param sz
 	 * @return log2
 	 */
 	static int log(long sz) {
 		long base = 2;
 		if (sz <= 0) {
 			throw new IllegalArgumentException("log2 negative"); //$NON-NLS-1$
 		}
 		int l = 0;
 		while (sz > 0) {
 			l++;
 			sz /= base;
 		}
 
 		return l - 1;
 	}
 
 	/**
 	 * A segment is a consecutive list of reftables of the same approximate
 	 * size.
 	 */
 	static class Segment {
 		// the approximate log_2 of the size.
 		int log;
 
 		// The total bytes in this segment
 		long bytes;
 
 		int start;
 
 		int end; // exclusive.
 
 		int size() {
 			return end - start;
 		}
 
 		Segment(int start, int end, int log, long bytes) {
 			this.log = log;
 			this.start = start;
 			this.end = end;
 			this.bytes = bytes;
 		}
 
 		Segment() {
 			this(0, 0, 0, 0);
 		}
 
 		@Override
 		public int hashCode() {
 			return 0; // appease error-prone
 		}
 
 		@Override
 		public boolean equals(Object other) {
 			Segment o = (Segment) other;
 			return o.bytes == bytes && o.log == log && o.start == start
 					&& o.end == end;
 		}
 
 		@SuppressWarnings("boxing")
 		@Override
 		public String toString() {
 			return String.format("{ [%d,%d) l=%d sz=%d }", start, end, log, //$NON-NLS-1$
 					bytes);
 		}
 	}
 
 	static List<Segment> segmentSizes(long sizes[]) {
 		List<Segment> segments = new ArrayList<>();
 		Segment cur = new Segment();
 		for (int i = 0; i < sizes.length; i++) {
 			int l = log(sizes[i]);
 			if (l != cur.log && cur.bytes > 0) {
 				segments.add(cur);
 				cur = new Segment();
 				cur.start = i;
 				cur.log = l;
 			}
 
 			cur.log = l;
 			cur.end = i + 1;
 			cur.bytes += sizes[i];
 		}
 		segments.add(cur);
 		return segments;
 	}
 
 	private static Optional<Segment> autoCompactCandidate(long sizes[]) {
 		if (sizes.length == 0) {
 			return Optional.empty();
 		}
 
 		// The cost of compaction is proportional to the size, and we want to
 		// avoid frequent large compactions. We do this by playing the game 2048
 		// here: first compact together the smallest tables if there are more
 		// than one. Then try to see if the result will be big enough to match
 		// up with next up.
 
 		List<Segment> segments = segmentSizes(sizes);
 		segments = segments.stream().filter(s -> s.size() > 1)
 				.collect(Collectors.toList());
 		if (segments.isEmpty()) {
 			return Optional.empty();
 		}
 
 		Optional<Segment> optMinSeg = segments.stream()
 				.min(Comparator.comparing(s -> Integer.valueOf(s.log)));
 		// Input is non-empty, so always present.
 		Segment smallCollected = optMinSeg.get();
 		while (smallCollected.start > 0) {
 			int prev = smallCollected.start - 1;
 			long prevSize = sizes[prev];
 			if (log(smallCollected.bytes) < log(prevSize)) {
 				break;
 			}
 			smallCollected.start = prev;
 			smallCollected.bytes += prevSize;
 		}
 
 		return Optional.of(smallCollected);
 	}
 
 	/**
 	 * Heuristically tries to compact the stack if the stack has a suitable
 	 * shape.
 	 *
 	 * @throws IOException
 	 */
 	private void autoCompact() throws IOException {
 		Optional<Segment> cand = autoCompactCandidate(tableSizes());
 		if (cand.isPresent()) {
 			if (!compactRange(cand.get().start, cand.get().end - 1)) {
 				stats.failed++;
 			}
 		}
 	}
 
 	// 68b footer, 24b header = 92.
 	private static long OVERHEAD = 91;
 
 	private long[] tableSizes() throws IOException {
 		long[] sizes = new long[stack.size()];
 		for (int i = 0; i < stack.size(); i++) {
 			// If we don't subtract the overhead, the file size isn't
 			// proportional to the number of entries. This will cause us to
 			// compact too often, which is expensive.
 			sizes[i] = stack.get(i).reftableReader.size() - OVERHEAD;
 		}
 		return sizes;
 	}
 
 	void compactFully() throws IOException {
 		if (!compactRange(0, stack.size() - 1)) {
 			stats.failed++;
 		}
 	}
 }