package org.embulk.output.sftp;

import com.google.common.base.Throwables;
import org.apache.commons.vfs2.FileObject;
import org.apache.commons.vfs2.FileSystemException;
import org.apache.commons.vfs2.FileSystemOptions;
import org.apache.commons.vfs2.Selectors;
import org.apache.commons.vfs2.impl.StandardFileSystemManager;
import org.apache.commons.vfs2.provider.sftp.IdentityInfo;
import org.apache.commons.vfs2.provider.sftp.SftpFileSystemConfigBuilder;
import org.embulk.config.TaskReport;
import org.embulk.spi.Buffer;
import org.embulk.spi.Exec;
import org.embulk.spi.FileOutput;
import org.embulk.spi.TransactionalFileOutput;
import org.slf4j.Logger;

import java.io.File;
import java.io.IOException;
import java.io.OutputStream;
import java.net.URI;
import java.net.URISyntaxException;

import static org.embulk.output.sftp.SftpFileOutputPlugin.PluginTask;

/**
 * Created by takahiro.nakayama on 10/20/15.
 */
public class SftpFileOutput
    implements FileOutput, TransactionalFileOutput
{
    private final Logger logger = Exec.getLogger(SftpFileOutput.class);
    private final StandardFileSystemManager manager;
    private final FileSystemOptions fsOptions;
    private final String userInfo;
    private final String host;
    private final int port;
    private final String pathPrefix;
    private final String sequenceFormat;
    private final String fileNameExtension;

    private final int taskIndex;
    private int fileIndex = 0;
    private FileObject currentRamFile;

    private StandardFileSystemManager initializeStandardFileSystemManager()
    {
//        TODO: change logging format: org.apache.commons.logging.Log
//        System.setProperty("org.apache.commons.logging.Log",
//                           "org.apache.commons.logging.impl.NoOpLog");
        StandardFileSystemManager manager = new StandardFileSystemManager();
        try {
            manager.init();
        }
        catch (FileSystemException e) {
            logger.error(e.getMessage());
            throw new RuntimeException(e);
        }

        return manager;
    }

    private String initializeUserInfo(PluginTask task)
    {
        String userInfo = task.getUser();
        if (task.getPassword().isPresent()) {
            userInfo += ":" + task.getPassword().get();
        }
        return userInfo;
    }

    private FileSystemOptions initializeFsOptions(PluginTask task)
    {
        FileSystemOptions fsOptions = new FileSystemOptions();

        try {
            SftpFileSystemConfigBuilder.getInstance().setUserDirIsRoot(fsOptions, task.getUserDirIsRoot());
            SftpFileSystemConfigBuilder.getInstance().setStrictHostKeyChecking(fsOptions, "no");
            if (task.getSecretKeyFilePath().isPresent()) {
                IdentityInfo identityInfo = new IdentityInfo(new File((task.getSecretKeyFilePath().get())), task.getSecretKeyPassphrase().getBytes());
                SftpFileSystemConfigBuilder.getInstance().setIdentityInfo(fsOptions, identityInfo);
            }
        }
        catch (FileSystemException e) {
            logger.error(e.getMessage());
            throw new RuntimeException(e);
        }

        return fsOptions;
    }

    SftpFileOutput(PluginTask task, int taskIndex)
    {
        this.manager = initializeStandardFileSystemManager();
        this.userInfo = initializeUserInfo(task);
        this.fsOptions = initializeFsOptions(task);
        this.host = task.getHost();
        this.port = task.getPort();
        this.pathPrefix = task.getPathPrefix();
        this.sequenceFormat = task.getSequenceFormat();
        this.fileNameExtension = task.getFileNameExtension();
        this.taskIndex = taskIndex;
    }

    @Override
    public void nextFile()
    {
        closeCurrentWithUpload();

        try {
            currentRamFile = newRamFile(getRamUri(getOutputFilePath()));
            logger.info("new ram file: {}", currentRamFile.getPublicURIString());
        }
        catch (FileSystemException e) {
            logger.error(e.getMessage());
            Throwables.propagate(e);
        }
        catch (URISyntaxException e) {
            logger.error(e.getMessage());
            Throwables.propagate(e);
        }
    }

    @Override
    public void add(Buffer buffer)
    {
        if (currentRamFile == null) {
            throw new IllegalStateException("nextFile() must be called before poll()");
        }

        try {
            getCurrentRamFileOutputStream().write(buffer.array(), buffer.offset(), buffer.limit());
        }
        catch (IOException e) {
            logger.error(e.getMessage());
            Throwables.propagate(e);
        }
        buffer.release();
    }

    @Override
    public void finish()
    {
        closeCurrentWithUpload();
    }

    @Override
    public void close()
    {
        closeCurrentWithUpload();
        manager.close();
    }

    @Override
    public void abort()
    {
    }

    @Override
    public TaskReport commit()
    {
        return null;
    }

    private void closeCurrentWithUpload()
    {
        try {
            closeCurrentRamFileContent();
            uploadCurrentRamFileToSftp();
            closeCurrentRamFile();
        }
        catch (FileSystemException e) {
            logger.error(e.getMessage());
            Throwables.propagate(e);
        }
        catch (URISyntaxException e) {
            logger.error(e.getMessage());
            Throwables.propagate(e);
        }
        fileIndex++;
        currentRamFile = null;
    }

    private void closeCurrentRamFileContent()
            throws FileSystemException
    {
        if (currentRamFile == null) {
            return;
        }
        currentRamFile.getContent().close();
    }

    private void uploadCurrentRamFileToSftp()
            throws FileSystemException, URISyntaxException
    {
        if (currentRamFile == null) {
            return;
        }

        try (FileObject remoteSftpFile = newSftpFile(getSftpUri(getOutputFilePath()))) {
            remoteSftpFile.copyFrom(currentRamFile, Selectors.SELECT_SELF);
            logger.info("Upload: {}", remoteSftpFile.getPublicURIString());
        }
    }

    private void closeCurrentRamFile()
            throws FileSystemException
    {
        if (currentRamFile == null) {
            return;
        }

        currentRamFile.close();
        currentRamFile.delete();
    }

    private URI getSftpUri(String remoteFilePath)
            throws URISyntaxException
    {
        return new URI("sftp", userInfo, host, port, remoteFilePath, null, null);
    }

    private URI getRamUri(String remoteFilePath)
            throws URISyntaxException
    {
        return new URI("ram", null, remoteFilePath, null);
    }

    private String getOutputFilePath()
    {
        return pathPrefix + String.format(sequenceFormat, taskIndex, fileIndex) + fileNameExtension;
    }

    private FileObject newSftpFile(URI sftpUri)
            throws FileSystemException
    {
        return manager.resolveFile(sftpUri.toString(), fsOptions);
    }

    private FileObject newRamFile(URI ramUri)
            throws FileSystemException
    {
        FileObject ramFile = manager.resolveFile(ramUri);
        ramFile.createFile();
        return ramFile;
    }

    private OutputStream getCurrentRamFileOutputStream()
            throws FileSystemException
    {
        return currentRamFile.getContent().getOutputStream(true);
    }
}

@@ -1485,16 +1485,69 @@
 RamFile;
+%0A    private OutputStream currentRamFileOutputStream;
 %0A%0A    pr
@@ -3897,24 +3897,98 @@
 lePath()));%0A
+            currentRamFileOutputStream = getCurrentRamFileOutputStream();%0A
             
@@ -4049,32 +4049,32 @@
 icURIString());%0A
-
         %7D%0A      
@@ -4549,20 +4549,17 @@
         
-getC
+c
 urrentRa
@@ -4575,18 +4575,16 @@
 utStream
-()
 .write(b
@@ -5324,34 +5324,33 @@
      catch (
-FileSystem
+URISyntax
 Exception e)
@@ -5452,33 +5452,26 @@
      catch (
-URISyntax
+IO
 Exception e)
@@ -5682,34 +5682,26 @@
      throws 
-FileSystem
+IO
 Exception%0A  
@@ -5754,32 +5754,32 @@
         return;%0A
-
         %7D%0A      
@@ -5764,32 +5764,76 @@
 turn;%0A        %7D%0A
+        currentRamFileOutputStream.close();%0A
         currentR

fix a bug: Could not write to "ram:///tmp/hoge" because it is currently in use.