package io.delphiplatform.api.extapi;

import org.apache.hc.core5.http.ParseException;
import org.apache.hc.core5.http.message.BasicHeader;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.HttpStatus;
import org.springframework.stereotype.Service;
import org.springframework.web.server.ResponseStatusException;

import java.io.IOException;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

import javax.annotation.PostConstruct;

import io.delphiplatform.api.util.FunctionUtils;
import lombok.extern.slf4j.Slf4j;
import se.michaelthelin.spotify.SpotifyApi;
import se.michaelthelin.spotify.exceptions.SpotifyWebApiException;
import se.michaelthelin.spotify.exceptions.detailed.BadGatewayException;
import se.michaelthelin.spotify.exceptions.detailed.BadRequestException;
import se.michaelthelin.spotify.exceptions.detailed.InternalServerErrorException;
import se.michaelthelin.spotify.exceptions.detailed.NotFoundException;
import se.michaelthelin.spotify.exceptions.detailed.ServiceUnavailableException;
import se.michaelthelin.spotify.exceptions.detailed.UnauthorizedException;
import se.michaelthelin.spotify.model_objects.credentials.ClientCredentials;
import se.michaelthelin.spotify.model_objects.specification.Album;
import se.michaelthelin.spotify.model_objects.specification.Paging;
import se.michaelthelin.spotify.model_objects.specification.Playlist;
import se.michaelthelin.spotify.model_objects.specification.PlaylistTrack;
import se.michaelthelin.spotify.requests.IRequest;
import se.michaelthelin.spotify.requests.authorization.client_credentials.ClientCredentialsRequest;
import se.michaelthelin.spotify.requests.data.AbstractDataPagingRequest;
import se.michaelthelin.spotify.requests.data.AbstractDataRequest;

import static com.google.common.net.HttpHeaders.AUTHORIZATION;

@Slf4j
@Service
public class SpotifyApiRequestExecutor {

    private static final int API_CALL_MAX_RETRIES = 3;
    private static final long API_CALL_RETRY_DELAY_MILLS = 3000;

    @Value("${delphi.spotify.client-id}")
    private String clientId;
    @Value("${delphi.spotify.client-secret}")
    private String clientSecret;

    private SpotifyApi spotifyApi;
    private final AtomicLong validTo;

    public SpotifyApiRequestExecutor() {
        this.validTo = new AtomicLong(0);
    }

    protected SpotifyApiRequestExecutor(SpotifyApi spotifyApi) {
        this();
        this.spotifyApi = spotifyApi;
    }

    @PostConstruct
    private void init() {
        this.spotifyApi = new SpotifyApi.Builder()
            .setClientId(clientId)
            .setClientSecret(clientSecret)
            .build();
    }

    public Album[] getSeveralAlbums(String... albumIds) {
        return executeForOne(api -> api.getSeveralAlbums(albumIds));
    }

    public Playlist getPlaylist(String playlistId) {
        return executeForOne(api -> api.getPlaylist(playlistId));
    }

    public <T, BT extends AbstractDataRequest.Builder<T, ?>> T executeForOne(
        Function<SpotifyApi, AbstractDataRequest.Builder<T, BT>> operation) {
        AbstractDataRequest.Builder<T, BT> requestBuilder = operation.apply(spotifyApi);
        return queryForOne(requestBuilder);
    }

    private <T, BT extends AbstractDataRequest.Builder<T, ?>> T queryForOne(AbstractDataRequest.Builder<T, BT> requestBuilder) {
        try {
            return executeAuthenticated(requestBuilder.build());
        } catch (NotFoundException e) {
            log.debug("The item requested could not be found.", e);
        }
        return null;
    }

    public List<PlaylistTrack> getPlaylistsItems(String playlistId) {
        return executeForList(api -> api.getPlaylistsItems(playlistId));
    }

    public <T, BT extends AbstractDataPagingRequest.Builder<T, ?>> List<T> executeForList(
        Function<SpotifyApi, AbstractDataPagingRequest.Builder<T, BT>> operation) {
        AbstractDataPagingRequest.Builder<T, BT> requestBuilder = operation.apply(spotifyApi);
        return queryForList(requestBuilder);
    }

    private <T, BT extends AbstractDataPagingRequest.Builder<T, ?>> List<T> queryForList(
        AbstractDataPagingRequest.Builder<T, BT> requestBuilder) {
        try {
            Paging<T> pagedResponse = executeAuthenticated(requestBuilder.build());
            List<T> items = Arrays.stream(pagedResponse.getItems())
                .collect(Collectors.toList());

            int pageCount = getPageCount(pagedResponse);
            int limit = pagedResponse.getLimit();

            if (pageCount > 1) {
                List<CompletableFuture<Paging<T>>> completableFutureList = IntStream.range(1, pageCount)
                    .mapToObj(i -> requestBuilder
                        .offset(i * limit)
                        .limit(limit)
                        .build()
                        .executeAsync())
                    .collect(Collectors.toList());

                combineAsyncPagingWithInitial(items, completableFutureList);
            }
            return items;
        } catch (NotFoundException e) {
            return Collections.emptyList();
        }
    }

    protected <T> T executeAuthenticated(IRequest<T> request) throws NotFoundException {
        int attempt = 1;

        while (true) {
            try {
                return doExecuteAuthenticated(request);
            } catch (InternalServerErrorException | BadGatewayException | ServiceUnavailableException e) {
                log.debug(String.format("Failed to query Spotify API. Attempt #%s", attempt), e);

                if (attempt >= API_CALL_MAX_RETRIES) {
                    throw new RuntimeException(String.format("Spotify API request error (after %s re-tries)", API_CALL_MAX_RETRIES), e);
                }

                attempt++;
                FunctionUtils.threadSleep(API_CALL_RETRY_DELAY_MILLS);
            } catch (NotFoundException e) {
                throw e; //may be processed differently by calling code
            } catch (BadRequestException e) {
                throw new ResponseStatusException(HttpStatus.BAD_REQUEST, e.getMessage(), e);
            } catch (SpotifyWebApiException | IOException | ParseException e) {
                throw new RuntimeException(String.format("Spotify API request error: %s", e.getMessage()), e);
            }
        }
    }

    protected <T> T doExecuteAuthenticated(IRequest<T> request) throws SpotifyWebApiException, IOException, ParseException {
        log.info("Spotify API request: {}", request.getUri().getPath());
        if (Instant.now().getEpochSecond() >= validTo.get()) {
            authenticate(request);
        }

        try {
            return request.execute();
        } catch (UnauthorizedException e) {
            authenticate(request);

            return request.execute();
        }
    }

    private <T> void authenticate(IRequest<T> request) throws IOException, ParseException, SpotifyWebApiException {
        log.info("Spotify API client credentials request");

        ClientCredentialsRequest clientCredentialsRequest = spotifyApi.clientCredentials().build();
        ClientCredentials clientCredentials = clientCredentialsRequest.execute();
        synchronized (this) {
            validTo.set(Instant.now().getEpochSecond() + clientCredentials.getExpiresIn());
            String accessToken = clientCredentials.getAccessToken();
            spotifyApi.setAccessToken(accessToken);

            if (request != null) {
                request.getHeaders().removeIf(header -> AUTHORIZATION.equals(header.getName()));
                request.getHeaders().add(new BasicHeader(AUTHORIZATION, "Bearer " + accessToken));
            }
        }
    }

    private int getPageCount(Paging<?> paging) {
        int total = paging.getTotal();
        int limit = paging.getLimit();
        int pageCount = total / limit;
        if (total % limit != 0) {
            pageCount++;
        }
        return pageCount;
    }

    private <T> void combineAsyncPagingWithInitial(List<T> startList, List<CompletableFuture<Paging<T>>> futures) {
        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();

        startList.addAll(
            futures.stream()
                .map(CompletableFuture::join)
                .map(Paging::getItems)
                .collect(ArrayList::new,
                    (accum, items) -> accum.addAll(Arrays.asList(items)),
                    ArrayList::addAll)
        );
    }
}