package com.atlassian.bitbucket.internal.scm.git.lfs.mirror;

import com.atlassian.bitbucket.internal.scm.git.lfs.embedded.ObjectStreamingOutput;
import com.atlassian.bitbucket.internal.scm.git.lfs.rest.storage.HttpRange;
import com.atlassian.bitbucket.repository.Repository;
import com.atlassian.bitbucket.throttle.ThrottleService;
import com.atlassian.bitbucket.throttle.Ticket;
import com.atlassian.util.contentcache.CacheAccess;
import com.atlassian.util.contentcache.ContentCache;

import javax.annotation.Nonnull;
import javax.ws.rs.WebApplicationException;
import javax.ws.rs.core.StreamingOutput;
import java.io.IOException;
import java.io.OutputStream;
import java.util.Optional;

import static com.atlassian.bitbucket.internal.scm.git.lfs.mirror.LfsCacheKey.lfsCacheKey;
import static java.util.Objects.requireNonNull;

/**
 * An implementation of {@link StreamingOutput} that streams data from a content-cache to the output stream.
 * This implementation holds onto a LFS hosting {@link Ticket} while the operation is in progress.
 */
class LfsCachedStreamingOutput implements ObjectStreamingOutput {

    private final ContentCache cache;
    private final LfsCacheKey cacheKey;
    private final UpstreamLfsClient client;
    private final Repository repository;
    private final String resourceName;
    private final ThrottleService throttleService;

    public LfsCachedStreamingOutput(@Nonnull ContentCache cache, @Nonnull UpstreamLfsClient client,
                                    @Nonnull String oid, @Nonnull Repository repository,
                                    @Nonnull ThrottleService throttleService, @Nonnull String resourceName) {
        cacheKey = requireNonNull(lfsCacheKey(repository, oid), "lfsCacheKey");
        this.cache = requireNonNull(cache, "cache");
        this.client = requireNonNull(client, "client");
        this.resourceName = requireNonNull(resourceName, "resourceName");
        this.repository = requireNonNull(repository, "repository");
        this.throttleService = requireNonNull(throttleService, "throttleService");
    }

    /**
     * Always returns {@link Optional#empty()} as byte ranges are not supported by this class
     */
    @Nonnull
    @Override
    public Optional<HttpRange> range() {
        return Optional.empty();
    }

    @Override
    public void write(OutputStream output) throws IOException, WebApplicationException {
        try (Ticket ignored = throttleService.acquireTicket(resourceName);
             CacheAccess cacheAccess = cache.access(cacheKey.cacheKey(), output,
                     new LfsContentProvider(repository, cacheKey.oid(), client))
        ) {
            cacheAccess.stream();
        }
    }
}
