/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You 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 org.apache.activemq.artemis.jdbc.store.file;

import java.io.File;
import java.io.IOException;
import java.lang.invoke.MethodHandles;
import java.nio.ByteBuffer;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.api.core.ActiveMQBuffers;
import org.apache.activemq.artemis.api.core.ActiveMQException;
import org.apache.activemq.artemis.api.core.ActiveMQExceptionType;
import org.apache.activemq.artemis.api.core.ActiveMQIOErrorException;
import org.apache.activemq.artemis.core.io.IOCallback;
import org.apache.activemq.artemis.core.io.SequentialFile;
import org.apache.activemq.artemis.core.io.buffer.TimedBuffer;
import org.apache.activemq.artemis.core.journal.EncodingSupport;
import org.apache.activemq.artemis.core.journal.impl.SimpleWaitIOCallback;
import org.apache.activemq.artemis.core.server.ActiveMQScheduledComponent;
import org.apache.activemq.artemis.utils.ReusableLatch;
import org.jctools.queues.MpscUnboundedArrayQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class JDBCSequentialFile implements SequentialFile {

   private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());

   private final String filename;

   private final String extension;

   private AtomicBoolean isOpen = new AtomicBoolean(false);

   private AtomicBoolean isLoaded = new AtomicBoolean(false);

   private long id = -1;

   private long readPosition = 0;

   private long writePosition = 0;

   private final Executor executor;

   private final JDBCSequentialFileFactory fileFactory;

   private final JDBCSequentialFileFactoryDriver dbDriver;

   MpscUnboundedArrayQueue<ScheduledWrite> writeQueue = new MpscUnboundedArrayQueue<>(8192);

   // Allows DB Drivers to cache meta data.
   private final Map<Object, Object> metaData = new ConcurrentHashMap<>();

   final JDBCPageWriteScheduler pageWriteScheduler;

   final ScheduledExecutorService scheduledExecutorService;

   private final ReusableLatch pendingWrites = new ReusableLatch();

   final long syncDelay;

   JDBCSequentialFile(final JDBCSequentialFileFactory fileFactory,
                      final String filename,
                      final Executor executor,
                      final ScheduledExecutorService scheduledExecutorService,
                      final long syncDelay,
                      final JDBCSequentialFileFactoryDriver driver) throws SQLException {
      this.fileFactory = fileFactory;
      this.filename = filename;
      this.extension = filename.contains(".") ? filename.substring(filename.lastIndexOf(".") + 1, filename.length()) : "";
      this.executor = executor;
      this.dbDriver = driver;
      this.scheduledExecutorService = scheduledExecutorService;
      this.syncDelay = syncDelay;
      this.pageWriteScheduler = new JDBCPageWriteScheduler(scheduledExecutorService, executor, syncDelay);
   }

   void setWritePosition(long writePosition) {
      this.writePosition = writePosition;
   }

   @Override
   public boolean isOpen() {
      return isOpen.get();
   }

   @Override
   public boolean exists() {
      try {
         return dbDriver.getFileID(this) >= 0;
      } catch (Throwable e) {
         logger.warn(e.getMessage(), e);
         return false;
      }
   }

   @Override
   public void open() throws Exception {
      isOpen.compareAndSet(false, load());
   }

   private boolean load() {
      try {
         if (isLoaded.compareAndSet(false, true)) {
            dbDriver.openFile(this);
         }
         return true;
      } catch (SQLException e) {
         isLoaded.set(false);
         // should not throw exceptions, as we drop the table on queue.destroy.
         // storage.exists could be called for non existing pages during async cleanup and they are
         // just supposed to return false
      }
      return false;
   }

   @Override
   public void open(int maxIO, boolean useExecutor) throws Exception {
      open();
   }

   @Override
   public boolean fits(int size) {
      return writePosition + size <= dbDriver.getMaxSize();
   }

   @Override
   public int calculateBlockStart(int position) throws Exception {
      return 0;
   }

   @Override
   public String getFileName() {
      return filename;
   }

   @Override
   public void fill(int size) throws Exception {
      // Do nothing
   }

   @Override
   public ByteBuffer map(int position, long size) throws IOException {
      return null;
   }

   @Override
   public void delete() throws IOException, InterruptedException, ActiveMQException {
      try {
         synchronized (this) {
            if (load()) {
               dbDriver.deleteFile(this);
            }
         }
      } catch (SQLException e) {
         // file is already gone from a drop somewhere
         logger.debug("Expected error deleting Sequential File", e);
         return;
      }
   }

   private synchronized int jdbcWrite(byte[] data, IOCallback callback, boolean append) {
      try {
         logger.debug("Writing {} bytes into {}", data.length, filename);
         synchronized (this) {
            int noBytes = dbDriver.writeToFile(this, data, append);
            seek(append ? writePosition + noBytes : noBytes);
            if (logger.isTraceEnabled()) {
               logger.trace("Write: ID: {} FileName: {}{}", getId(), getFileName(), size());
            }
            if (callback != null)
               callback.done();
            return noBytes;
         }
      } catch (Exception e) {
         if (callback != null)
            callback.onError(ActiveMQExceptionType.IO_ERROR.getCode(), e.getMessage());
         fileFactory.onIOError(e, "Error writing to JDBC file.", this);
      }
      return 0;
   }

   public synchronized int jdbcWrite(ActiveMQBuffer buffer, IOCallback callback) {
      return jdbcWrite(buffer, callback, true);
   }

   public synchronized int jdbcWrite(ActiveMQBuffer buffer, IOCallback callback, boolean append) {
      byte[] data = new byte[buffer.readableBytes()];
      buffer.readBytes(data);
      return jdbcWrite(data, callback, append);
   }



   private void pollWrites() {
      if (writeQueue.isEmpty()) {
         return;
      }

      logger.debug("polling {} elements on {}", writeQueue.size(), this.filename);

      ArrayList<ScheduledWrite> writeList = new ArrayList<>(writeQueue.size()); // the size here is just an estimate

      byte[] bytes = extractBytes(writeList);

      jdbcWrite(bytes, null, true);
      writeList.forEach(this::doCallback);
   }

   /* Even though I would love to have a reusable byte[] for the following buffer
      PreparedStatement.setData takes a byte[] without any sizing on the interface.
      Blob interface would support setBytes with an offset and size, but some of the databases we are using
      (DB2 specifically) is not allowing us to use Blob (at least during our dev time).
      for that reason I'm using this byte[] with the very specific size that needs to be written

      Also Notice that our PagingManager will make sure that this size wouldn't go beyond our page-size limit
      which we also limit at the JDBC storage, which should be 100K. */
   private byte[] extractBytes(ArrayList<ScheduledWrite> writeList) {
      int totalSize = 0;
      ScheduledWrite write;
      while ((write = writeQueue.poll()) != null) {
         writeList.add(write);
         totalSize += write.readable();
      }
      byte[] bytes = new byte[totalSize];

      int writePosition = 0;

      for (ScheduledWrite el : writeList) {
         writePosition += el.readAt(bytes, writePosition);
         el.releaseBuffer();
      }

      return bytes;
   }

   private void doCallback(ScheduledWrite write) {
      if (write != null && write.callback != null) {
         write.callback.done();
      }
      pendingWrites.countDown();
   }

   private void scheduleWrite(final ActiveMQBuffer bytes, final IOCallback callback, boolean append) {
      scheduleWrite(new ScheduledWrite(bytes, callback, append));
   }

   private void scheduleWrite(ScheduledWrite scheduledWrite) {
      logger.debug("offering {} bytes into {}", scheduledWrite.readable(), filename);
      pendingWrites.countUp();
      writeQueue.offer(scheduledWrite);
      this.pageWriteScheduler.delay();
   }

   private void scheduleWrite(final ByteBuffer bytes, final IOCallback callback) {
      scheduleWrite(new ScheduledWrite(bytes, callback, true));
   }

   synchronized void seek(long noBytes) {
      writePosition = noBytes;
   }

   public void sendToDB(ActiveMQBuffer bytes, IOCallback callback, boolean append) throws Exception {
      SimpleWaitIOCallback waitIOCallback = null;

      if (callback == null) {
         waitIOCallback = new SimpleWaitIOCallback();
         callback = waitIOCallback;
      }

      scheduleWrite(bytes, callback, append);

      if (callback != null) {
         waitIOCallback.waitCompletion();
      }
   }

   @Override
   public void write(ActiveMQBuffer bytes, boolean sync, IOCallback callback) throws Exception {
      sendToDB(bytes, callback, true);
   }

   @Override
   public void write(ActiveMQBuffer bytes, boolean sync) throws Exception {
      write(bytes, sync, null);
   }

   @Override
   public void write(EncodingSupport bytes, boolean sync, IOCallback callback) throws Exception {
      ActiveMQBuffer data = ActiveMQBuffers.fixedBuffer(bytes.getEncodeSize());
      bytes.encode(data);
      sendToDB(data, callback, true);
   }

   @Override
   public void write(EncodingSupport bytes, boolean sync) throws Exception {
      write(bytes, sync, null);
   }

   @Override
   public void writeDirect(ByteBuffer bytes, boolean sync, IOCallback callback) {
      if (callback == null) {
         final SimpleWaitIOCallback waitIOCallback = sync ? new SimpleWaitIOCallback() : null;
         try {
            scheduleWrite(bytes, waitIOCallback);
            if (waitIOCallback != null) {
               waitIOCallback.waitCompletion();
            }
         } catch (Exception e) {
            waitIOCallback.onError(ActiveMQExceptionType.IO_ERROR.getCode(), "Error writing to JDBC file.");
            fileFactory.onIOError(e, "Failed to write to file.", this);
         }
      } else {
         scheduleWrite(bytes, callback);
      }

   }

   @Override
   public void blockingWriteDirect(ByteBuffer bytes, boolean sync, boolean releaseBuffer) {
      writeDirect(bytes, sync, null);
   }

   @Override
   public void writeDirect(ByteBuffer bytes, boolean sync) throws Exception {
      writeDirect(bytes, sync, null);
      // Are we meant to block here?
   }

   @Override
   public synchronized int read(ByteBuffer bytes, final IOCallback callback) throws SQLException {
      synchronized (this) {
         try {
            int read = dbDriver.readFromFile(this, bytes);
            readPosition += read;
            if (callback != null)
               callback.done();
            return read;
         } catch (SQLException e) {
            if (callback != null)
               callback.onError(ActiveMQExceptionType.IO_ERROR.getCode(), e.getMessage());
            fileFactory.onIOError(e, "Error reading from JDBC file.", this);
         }
         return 0;
      }
   }

   @Override
   public int read(ByteBuffer bytes) throws Exception {
      return read(bytes, null);
   }

   @Override
   public void position(long pos) throws IOException {
      readPosition = pos;
   }

   @Override
   public long position() {
      return readPosition;
   }


   @Override
   public void close() throws Exception {
      close(true, true);
   }

   @Override
   public void close(boolean waitOnSync, boolean block) throws Exception {
      isOpen.set(false);
      if (waitOnSync) {
         sync();
      }
      fileFactory.sequentialFileClosed(this);
   }

   public int getNetworkTimeoutMillis() {
      return dbDriver.getJdbcConnectionProvider().getNetworkTimeoutMillis();
   }

   @Override
   public void sync() throws IOException {
      try {
         int syncTimeout = getNetworkTimeoutMillis();

         if (syncTimeout >= 0) {
            if (!pendingWrites.await(syncTimeout, TimeUnit.MILLISECONDS)) {
               fileFactory.onIOError(new ActiveMQIOErrorException("Database not responding to syncs before timeout"), "Error during JDBC file sync.", this);
            }
         } else {
            // waiting forever however logger.debug while doing so
            while (!pendingWrites.await(1, TimeUnit.SECONDS)) {
               logger.debug("Awaiting syncs from database for page file {}", this.filename);
            }
         }

      } catch (Exception e) {
         fileFactory.onIOError(e, "Error during JDBC file sync.", this);
      }
   }

   @Override
   public long size() throws Exception {
      load();
      return writePosition;
   }

   @Override
   public void renameTo(String newFileName) throws Exception {
      synchronized (this) {
         try {
            dbDriver.renameFile(this, newFileName);
         } catch (SQLException e) {
            fileFactory.onIOError(e, "Error renaming JDBC file.", this);
         }
      }
   }

   @Override
   public SequentialFile cloneFile() {
      try {
         JDBCSequentialFile clone = new JDBCSequentialFile(fileFactory, filename, executor, scheduledExecutorService, syncDelay, dbDriver);
         clone.setWritePosition(this.writePosition);
         return clone;
      } catch (Exception e) {
         fileFactory.onIOError(e, "Error cloning JDBC file.", this);
      }
      return null;
   }

   @Override
   public void copyTo(SequentialFile cloneFile) throws Exception {
      JDBCSequentialFile clone = (JDBCSequentialFile) cloneFile;
      try {
         synchronized (this) {
            logger.trace("JDBC Copying File.  From: {} To: {}", this, cloneFile);
            clone.open();
            dbDriver.copyFileData(this, clone);
            clone.setWritePosition(writePosition);
         }
      } catch (Exception e) {
         fileFactory.onIOError(e, "Error copying JDBC file.", this);
      }
   }

   public long getId() {
      return id;
   }

   public void setId(long id) {
      this.id = id;
   }

   public String getFilename() {
      return filename;
   }

   public String getExtension() {
      return extension;
   }

   // Only Used by Journal, no need to implement.
   @Override
   public void setTimedBuffer(TimedBuffer buffer) {
   }

   // Only Used by replication, no need to implement.
   @Override
   public File getJavaFile() {
      return null;
   }

   public void addMetaData(Object key, Object value) {
      metaData.put(key, value);
   }

   public Object getMetaData(Object key) {
      return metaData.get(key);
   }



   private class JDBCPageWriteScheduler extends ActiveMQScheduledComponent {

      JDBCPageWriteScheduler(ScheduledExecutorService scheduledExecutorService,
                   Executor executor,
                   long checkPeriod) {
         super(scheduledExecutorService, executor, checkPeriod, TimeUnit.MILLISECONDS, true);
      }

      @Override
      public void run() {
         pollWrites();
      }
   }
}
