/*
 * 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.lang.invoke.MethodHandles;
import java.nio.ByteBuffer;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;

import org.apache.activemq.artemis.jdbc.store.drivers.JDBCConnectionProvider;
import org.apache.activemq.artemis.jdbc.store.sql.SQLProvider;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@SuppressWarnings("SynchronizeOnNonFinalField")
public final class PostgresSequentialSequentialFileDriver extends JDBCSequentialFileFactoryDriver {

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

   private static final String POSTGRES_OID_KEY = "POSTGRES_OID_KEY";
   private PostgresLargeObjectManager largeObjectManager;

   public PostgresSequentialSequentialFileDriver() throws SQLException {
      super();
   }

   public PostgresSequentialSequentialFileDriver(JDBCConnectionProvider connectionProvider, SQLProvider provider) {
      super();
      this.setJdbcConnectionProvider(connectionProvider);
      this.setSqlProvider(provider);
   }

   @Override
   protected void prepareStatements() {
      this.largeObjectManager = new PostgresLargeObjectManager();
      this.deleteFile = sqlProvider.getDeleteFileSQL();
      this.createFile = sqlProvider.getInsertFileSQL();
      this.createFileAutogeneratedKeys = Statement.RETURN_GENERATED_KEYS;
      this.selectFileByFileName = sqlProvider.getSelectFileByFileName();
      this.copyFileRecord = sqlProvider.getCopyFileRecordByIdSQL();
      this.renameFile = sqlProvider.getUpdateFileNameByIdSQL();
      this.readLargeObject = sqlProvider.getReadLargeObjectSQL();
      this.appendToLargeObject = sqlProvider.getAppendToLargeObjectSQL();
      this.selectFileNamesByExtension = sqlProvider.getSelectFileNamesByExtensionSQL();
   }

   @Override
   public void createFile(JDBCSequentialFile file) throws SQLException {
      try (Connection connection = connectionProvider.getConnection()) {
         try (PreparedStatement createFile = connection.prepareStatement(this.createFile, this.createFileAutogeneratedKeys)) {
            connection.setAutoCommit(false);
            Long oid = largeObjectManager.createLO(connection);
            file.setOid(oid);

            if (logger.isDebugEnabled()) {
               logger.debug("Creating file {} with oid={}", file.getFileName(), oid);
            }

            createFile.setString(1, file.getFileName());
            createFile.setString(2, file.getExtension());
            createFile.setLong(3, oid);
            createFile.executeUpdate();

            try (ResultSet keys = createFile.getGeneratedKeys()) {
               keys.next();
               file.setId(keys.getLong(1));
            }
            connection.commit();
         } catch (SQLException e) {
            connection.rollback();
            throw e;
         }
      }
   }

   @Override
   public void loadFile(JDBCSequentialFile file) throws SQLException {
      try (Connection connection = connectionProvider.getConnection()) {
         try (PreparedStatement readLargeObject = connection.prepareStatement(this.readLargeObject)) {
            connection.setAutoCommit(false);
            readLargeObject.setLong(1, file.getId());

            try (ResultSet rs = readLargeObject.executeQuery()) {
               if (rs.next()) {
                  file.setWritePosition(getPostgresLargeObjectSize(file));
               }
               connection.commit();
            } catch (SQLException e) {
               connection.rollback();
               throw e;
            }
         }
      }
   }

   @Override
   public int writeToFile(JDBCSequentialFile file, byte[] data, boolean append) throws SQLException {
      try (Connection connection = connectionProvider.getConnection()) {
         Long oid = getOID(file);
         try {
            connection.setAutoCommit(false);
            Object largeObject = largeObjectManager.open(connection, oid, PostgresLargeObjectManager.WRITE);
            if (append) {
               largeObjectManager.seek(largeObject, largeObjectManager.size(largeObject));
            } else {
               largeObjectManager.truncate(largeObject, 0);
            }
            largeObjectManager.write(largeObject, data);
            largeObjectManager.close(largeObject);
            connection.commit();
         } catch (Exception e) {
            connection.rollback();
            throw e;
         }
         return data.length;
      }
   }

   @Override
   public int readFromFile(JDBCSequentialFile file, ByteBuffer bytes) throws SQLException {
      long oid = getOID(file);
      try (Connection connection = connectionProvider.getConnection()) {
         try {
            connection.setAutoCommit(false);
            Object largeObject = largeObjectManager.open(connection, oid, PostgresLargeObjectManager.READ);
            int readLength = (int) calculateReadLength(largeObjectManager.size(largeObject), bytes.remaining(), file.position());

            if (readLength > 0) {
               if (file.position() > 0) {
                  largeObjectManager.seek(largeObject, (int) file.position());
               }
               byte[] data = largeObjectManager.read(largeObject, readLength);
               bytes.put(data);
            }

            largeObjectManager.close(largeObject);
            connection.commit();

            return readLength;
         } catch (SQLException e) {
            connection.rollback();
            throw e;
         }
      }
   }

   private Long getOID(JDBCSequentialFile file) throws SQLException {
      if (file.getOid() != -1L) {
         return file.getOid();
      }
      Long oid = (Long) file.getMetaData(POSTGRES_OID_KEY);
      if (oid == null) {
         try (Connection connection = connectionProvider.getConnection()) {
            try (PreparedStatement readLargeObject = connection.prepareStatement(this.readLargeObject)) {
               connection.setAutoCommit(false);
               readLargeObject.setLong(1, file.getId());
               try (ResultSet rs = readLargeObject.executeQuery()) {
                  if (rs.next()) {
                     file.addMetaData(POSTGRES_OID_KEY, rs.getLong(1));
                  }
                  connection.commit();
               } catch (SQLException e) {
                  connection.rollback();
                  throw e;
               }
            }
         }
      }
      Long value = (Long) file.getMetaData(POSTGRES_OID_KEY);
      if (value != null) {
         file.setOid(value);
      }
      return value;
   }

   private int getPostgresLargeObjectSize(JDBCSequentialFile file) throws SQLException {
      int size = 0;
      Long oid = getOID(file);
      if (oid != null) {
         try (Connection connection = connectionProvider.getConnection()) {
            try {
               connection.setAutoCommit(false);
               Object largeObject = largeObjectManager.open(connection, oid, PostgresLargeObjectManager.READ);
               size = largeObjectManager.size(largeObject);
               largeObjectManager.close(largeObject);
               connection.commit();
            } catch (SQLException e) {
               connection.rollback();
               throw e;
            }
         }
      }
      return size;
   }

   @Override
   public void deleteFile(JDBCSequentialFile file) throws SQLException {
      Long oid = getOID(file);
      if (logger.isDebugEnabled()) {
         logger.debug("Deleting file {} with oid={}", file.getFileName(), oid);
      }
      if (oid != null) {
         try (Connection connection = connectionProvider.getConnection()) {
            largeObjectManager.deleteLO(connection, oid);
         } catch (Exception e) {
            logger.warn(e.getMessage(), e);
         }
      }
      super.deleteFile(file);
   }


}
