/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * license agreements; and to You under the Apache License, version 2.0:
 *
 *   https://www.apache.org/licenses/LICENSE-2.0
 *
 * This file is part of the Apache Pekko project, which was derived from Akka.
 */

/*
 * Copyright (C) since 2016 Lightbend Inc. <https://www.lightbend.com>
 */

package docs.javadsl;

import static org.apache.pekko.util.ByteString.emptyByteString;
import static org.hamcrest.CoreMatchers.*;
import static org.hamcrest.MatcherAssert.*;

import java.io.File;
import java.net.URISyntaxException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.time.Instant;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.TimeUnit;
import org.apache.pekko.Done;
import org.apache.pekko.NotUsed;
import org.apache.pekko.actor.ActorSystem;
import org.apache.pekko.japi.Pair;
import org.apache.pekko.stream.IOResult;
import org.apache.pekko.stream.Materializer;
import org.apache.pekko.stream.connectors.file.ArchiveMetadata;
import org.apache.pekko.stream.connectors.file.TarArchiveMetadata;
import org.apache.pekko.stream.connectors.file.ZipArchiveMetadata;
import org.apache.pekko.stream.connectors.file.javadsl.Archive;
import org.apache.pekko.stream.connectors.file.javadsl.Directory;
import org.apache.pekko.stream.connectors.testkit.javadsl.LogCapturingExtension;
import org.apache.pekko.stream.javadsl.FileIO;
import org.apache.pekko.stream.javadsl.Sink;
import org.apache.pekko.stream.javadsl.Source;
import org.apache.pekko.stream.testkit.javadsl.StreamTestKit;
import org.apache.pekko.testkit.javadsl.TestKit;
import org.apache.pekko.util.ByteString;
import org.junit.jupiter.api.*;
import org.junit.jupiter.api.extension.ExtendWith;

@ExtendWith(LogCapturingExtension.class)
public class ArchiveTest {

  private static ActorSystem system;

  @BeforeAll
  public static void beforeAll() throws Exception {
    system = ActorSystem.create();
  }

  @AfterAll
  public static void afterAll() throws Exception {
    TestKit.shutdownActorSystem(system);
  }

  private ArchiveHelper archiveHelper;

  @BeforeEach
  public void setup() throws Exception {
    archiveHelper = new ArchiveHelper();
  }

  @Test
  public void flowShouldCreateZIPArchive() throws Exception {
    ByteString fileContent1 = readFileAsByteString(getFileFromResource("pekko.svg"));
    ByteString fileContent2 = readFileAsByteString(getFileFromResource("foundation.svg"));

    Source<ByteString, NotUsed> source1 = toSource(fileContent1);
    Source<ByteString, NotUsed> source2 = toSource(fileContent2);

    /*
    // #sample-zip
    Source<ByteString, NotUsed> source1 = ...
    Source<ByteString, NotUsed> source2 = ...

    // #sample-zip
    */

    // #sample-zip
    Pair<ArchiveMetadata, Source<ByteString, NotUsed>> pair1 =
        Pair.create(ArchiveMetadata.create("pekko.svg"), source1);
    Pair<ArchiveMetadata, Source<ByteString, NotUsed>> pair2 =
        Pair.create(ArchiveMetadata.create("foundation.svg"), source2);

    Source<Pair<ArchiveMetadata, Source<ByteString, NotUsed>>, NotUsed> source =
        Source.from(List.of(pair1, pair2));

    Sink<ByteString, CompletionStage<IOResult>> fileSink = FileIO.toPath(Paths.get("logo.zip"));
    CompletionStage<IOResult> ioResult = source.via(Archive.zip()).runWith(fileSink, system);

    // #sample-zip

    ioResult.toCompletableFuture().get(3, TimeUnit.SECONDS);

    Map<String, ByteString> inputFiles =
        new HashMap<String, ByteString>() {
          {
            put("pekko.svg", fileContent1);
            put("foundation.svg", fileContent2);
          }
        };

    ByteString resultFileContent = readFileAsByteString(Paths.get("logo.zip"));
    Map<String, ByteString> unzip = archiveHelper.unzip(resultFileContent);

    assertThat(inputFiles, is(unzip));
    Path target = Files.createTempDirectory("pekko-connectors-zip-");

    // #sample-zip-read
    Archive.zipReader(Paths.get("logo.zip").toFile())
        .mapAsync(
            4,
            pair -> {
              ZipArchiveMetadata metadata = pair.first();
              Path targetFile = target.resolve(metadata.getName());
              targetFile.toFile().getParentFile().mkdirs(); // missing error handler
              Source<ByteString, NotUsed> fSource = pair.second();
              // create the target directory
              return fSource
                  .runWith(FileIO.toPath(targetFile), system)
                  .thenApply(io -> Done.done());
            });
    // #sample-zip-read

    // cleanup
    new File("logo.zip").delete();
  }

  @Test
  public void flowShouldCreateTARArchive() throws Exception {
    Path filePath1 = getFileFromResource("pekko.svg");
    Path filePath2 = getFileFromResource("foundation.svg");

    ByteString fileContent1 = readFileAsByteString(filePath1);
    ByteString fileContent2 = readFileAsByteString(filePath2);

    Source<ByteString, NotUsed> source1 = toSource(fileContent1);
    Long size1 = Files.size(filePath1);
    Source<ByteString, NotUsed> source2 = toSource(fileContent2);
    Long size2 = Files.size(filePath2);

    /*
    // #sample-tar
    Source<ByteString, NotUsed> source1 = ...
    Source<ByteString, NotUsed> source2 = ...
    Long size1 = ...
    Long size2 = ...

    // #sample-tar
    */

    Instant lastModification = Instant.now();
    // #sample-tar
    Pair<TarArchiveMetadata, Source<ByteString, NotUsed>> dir =
        Pair.create(TarArchiveMetadata.directory("subdir", lastModification), Source.empty());

    Pair<TarArchiveMetadata, Source<ByteString, NotUsed>> pair1 =
        Pair.create(
            TarArchiveMetadata.create("subdir", "pekko.svg", size1, lastModification), source1);
    Pair<TarArchiveMetadata, Source<ByteString, NotUsed>> pair2 =
        Pair.create(TarArchiveMetadata.create("foundation.svg", size2, lastModification), source2);

    Source<Pair<TarArchiveMetadata, Source<ByteString, NotUsed>>, NotUsed> source =
        Source.from(List.of(dir, pair1, pair2));

    Sink<ByteString, CompletionStage<IOResult>> fileSink = FileIO.toPath(Paths.get("logo.tar"));
    CompletionStage<IOResult> ioResult = source.via(Archive.tar()).runWith(fileSink, system);
    // #sample-tar

    // #sample-tar-gz
    Sink<ByteString, CompletionStage<IOResult>> fileSinkGz =
        FileIO.toPath(Paths.get("logo.tar.gz"));
    CompletionStage<IOResult> ioResultGz =
        source
            .via(Archive.tar().via(org.apache.pekko.stream.javadsl.Compression.gzip()))
            .runWith(fileSinkGz, system);
    // #sample-tar-gz

    ioResult.toCompletableFuture().get(3, TimeUnit.SECONDS);
    ioResultGz.toCompletableFuture().get(3, TimeUnit.SECONDS);

    // cleanup
    new File("logo.tar").delete();
    new File("logo.tar.gz").delete();
  }

  @Test
  public void tarReader() throws Exception {
    ByteString tenDigits = ByteString.fromString("1234567890");
    TarArchiveMetadata metadata1 = TarArchiveMetadata.directory("dir/");
    TarArchiveMetadata metadata2 = TarArchiveMetadata.create("dir/file1.txt", tenDigits.length());
    CompletionStage<ByteString> oneFileArchive =
        Source.from(
                List.of(
                    Pair.create(metadata1, Source.empty(ByteString.class)),
                    Pair.create(metadata2, Source.single(tenDigits))))
            .via(Archive.tar())
            .runWith(Sink.fold(ByteString.emptyByteString(), ByteString::concat), system);

    // #tar-reader
    Source<ByteString, NotUsed> bytesSource = // ???
        // #tar-reader
        Source.completionStage(oneFileArchive);
    Path target = Files.createTempDirectory("pekko-connectors-tar-");

    // #tar-reader
    CompletionStage<Done> tar =
        bytesSource
            .via(Archive.tarReader())
            .mapAsync(
                1,
                pair -> {
                  TarArchiveMetadata metadata = pair.first();
                  Path targetFile = target.resolve(metadata.filePath());
                  if (metadata.isDirectory()) {
                    return Source.single(targetFile)
                        .via(Directory.mkdirs())
                        .runWith(Sink.ignore(), system);
                  } else {
                    Source<ByteString, NotUsed> source = pair.second();
                    // create the target directory
                    return Source.single(targetFile.getParent())
                        .via(Directory.mkdirs())
                        .runWith(Sink.ignore(), system)
                        .thenCompose(
                            done ->
                                // stream the file contents to a local file
                                source
                                    .runWith(FileIO.toPath(targetFile), system)
                                    .thenApply(io -> Done.done()));
                  }
                })
            .runWith(Sink.ignore(), system);
    // #tar-reader
    assertThat(tar.toCompletableFuture().get(2, TimeUnit.SECONDS), is(Done.done()));
    File file = target.resolve("dir/file1.txt").toFile();
    assertThat(file.exists(), is(true));
  }

  @Test
  public void tarReaderWithEofBlocks() throws Exception {
    ByteString tenDigits = ByteString.fromString("1234567890");
    TarArchiveMetadata metadata1 = TarArchiveMetadata.create("file5.txt", tenDigits.length());
    CompletionStage<ByteString> oneFileArchive =
        Source.single(Pair.create(metadata1, Source.single(tenDigits)))
            .via(Archive.tar())
            // tar standard suggests two empty 512 byte blocks as EOF marker
            .concat(Source.single(ByteString.fromArray(new byte[1024])))
            .runWith(Sink.fold(ByteString.emptyByteString(), ByteString::concat), system);

    Source<ByteString, NotUsed> bytesSource = Source.completionStage(oneFileArchive);
    Path target = Files.createTempDirectory("pekko-connectors-tar-");

    CompletionStage<Done> tar =
        bytesSource
            .via(Archive.tarReader())
            .mapAsync(
                1,
                pair -> {
                  TarArchiveMetadata metadata = pair.first();
                  Source<ByteString, NotUsed> source = pair.second();
                  Path targetFile = target.resolve(metadata.filePath());
                  return source.runWith(FileIO.toPath(targetFile), system);
                })
            .runWith(Sink.ignore(), system);
    assertThat(tar.toCompletableFuture().get(2, TimeUnit.SECONDS), is(Done.done()));
    File file = target.resolve("file5.txt").toFile();
    assertThat(file.exists(), is(true));
  }

  @AfterEach
  public void tearDown() throws Exception {
    StreamTestKit.assertAllStagesStopped(Materializer.matFromSystem(system));
  }

  private Path getFileFromResource(String fileName) throws URISyntaxException {
    return Paths.get(getClass().getClassLoader().getResource(fileName).toURI());
  }

  private ByteString readFileAsByteString(Path filePath) throws Exception {
    final Sink<ByteString, CompletionStage<ByteString>> foldSink =
        Sink.fold(emptyByteString(), ByteString::concat);

    return FileIO.fromPath(filePath)
        .runWith(foldSink, system)
        .toCompletableFuture()
        .get(3, TimeUnit.SECONDS);
  }

  private Source<ByteString, NotUsed> toSource(ByteString bs) {
    return Source.single(bs);
  }
}
