package io.lakefs.storage;

import java.io.IOException;
import java.io.OutputStream;
import java.net.URISyntaxException;
import java.util.Objects;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import io.lakefs.LakeFSClient;
import io.lakefs.LakeFSFileSystem;
import io.lakefs.LakeFSLinker;
import io.lakefs.clients.api.ApiException;
import io.lakefs.clients.api.ObjectsApi;
import io.lakefs.clients.api.StagingApi;
import io.lakefs.clients.api.model.ObjectStats;
import io.lakefs.clients.api.model.StagingLocation;
import io.lakefs.utils.ObjectLocation;

public class SimpleStorageAccessStrategy implements StorageAccessStrategy {
    private PhysicalAddressTranslator physicalAddressTranslator;
    private LakeFSFileSystem lakeFSFileSystem;
    private LakeFSClient lfsClient;
    private Configuration conf;

    public SimpleStorageAccessStrategy(LakeFSFileSystem lakeFSFileSystem, LakeFSClient lfsClient,
            Configuration conf, PhysicalAddressTranslator physicalAddressTranslator) {
        this.lakeFSFileSystem = lakeFSFileSystem;
        this.lfsClient = lfsClient;
        this.conf = conf;
        this.physicalAddressTranslator = physicalAddressTranslator;
    }

    @Override
    public FSDataOutputStream createDataOutputStream(ObjectLocation objectLocation,
            CreateOutputStreamParams params) throws ApiException, IOException {
        StagingApi staging = lfsClient.getStagingApi();
        StagingLocation stagingLocation = staging.getPhysicalAddress(objectLocation.getRepository(),
                objectLocation.getRef(), objectLocation.getPath(), false);
        Path physicalPath;
        try {
            physicalPath = physicalAddressTranslator.translate(Objects.requireNonNull(stagingLocation.getPhysicalAddress()));
        } catch (URISyntaxException e) {
            throw new IOException("Failed to parse object phystical address", e);
        }
        FileSystem physicalFs = physicalPath.getFileSystem(conf);
        OutputStream physicalOut;
        if (params != null) {
            physicalOut = physicalFs.create(physicalPath, false, params.bufferSize,
                    physicalFs.getDefaultReplication(physicalPath), params.blockSize);
        } else {
            physicalOut = physicalFs.create(physicalPath);
        }
        MetadataClient metadataClient = new MetadataClient(physicalFs);
        LakeFSLinker linker = new LakeFSLinker(lakeFSFileSystem, lfsClient, objectLocation, stagingLocation);
        LinkOnCloseOutputStream out = new LinkOnCloseOutputStream(physicalPath.toUri(), metadataClient, physicalOut, linker);
        // TODO(ariels): add fs.FileSystem.Statistics here to keep track.
        return new FSDataOutputStream(out, null);
    }
    
    @Override
    public FSDataInputStream createDataInputStream(ObjectLocation objectLocation, int bufSize)
            throws ApiException, IOException {
        ObjectsApi objects = lfsClient.getObjectsApi();
        ObjectStats stats = objects.statObject(objectLocation.getRepository(),
                objectLocation.getRef(), objectLocation.getPath(), false, false);
        Path physicalPath;
        try {
            physicalPath = physicalAddressTranslator.translate(Objects.requireNonNull(stats.getPhysicalAddress()));
        } catch (URISyntaxException e) {
            throw new IOException("Failed to parse object physical address", e);
        }
        FileSystem physicalFs = physicalPath.getFileSystem(conf);
        return physicalFs.open(physicalPath, bufSize);
    }
}
