/*
 * The MIT License (MIT)
 *
 * Copyright (c) Open Application Platform Authors
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 */
package oap.logstream;

import lombok.extern.slf4j.Slf4j;
import oap.io.Files;
import oap.io.IoStreams;
import oap.metrics.Metrics;
import org.apache.commons.lang3.StringUtils;
import org.joda.time.DateTime;
import org.joda.time.DateTimeUtils;

import java.nio.file.Path;
import java.util.HashSet;
import java.util.Objects;
import java.util.Set;
import java.util.stream.Collectors;

import static oap.io.IoStreams.Encoding.PLAIN;

@Slf4j
public class Archiver implements Runnable {
   private final Path sourceDirectory;
   private final Path destinationDirectory;
   private final long safeInterval;
   private final IoStreams.Encoding encoding;
   private final String mask;
   private final int bucketsPerHour;
   private Set<String> hourBasedFiles = new HashSet<>();
   private Path hourBasedDestination;
   private int bufferSize = 1024 * 256 * 4 * 4;


   public Archiver( Path sourceDirectory, Path destinationDirectory, long safeInterval, String mask,
                    IoStreams.Encoding encoding, int bucketsPerHour, Set<String> hourBasedFiles,
                    Path hourBasedDestination) {
      this.sourceDirectory = sourceDirectory;
      this.destinationDirectory = destinationDirectory;
      this.safeInterval = safeInterval;
      this.mask = mask;
      this.encoding = encoding;
      this.bucketsPerHour = bucketsPerHour;
      this.hourBasedFiles = hourBasedFiles;

      if (!hourBasedFiles.isEmpty()) {
         this.hourBasedDestination = Objects.requireNonNull(hourBasedDestination, "hour based destination has to be " +
            "specified, if there are files configured");
      }
   }

   @Override
   public void run() {
      log.debug( "let's start packing of {} in {} into {}", mask, sourceDirectory, destinationDirectory );
      String timestamp = Timestamp.format( DateTime.now(), bucketsPerHour );

      log.debug( "current timestamp is {}", timestamp );
      long elapsed = DateTimeUtils.currentTimeMillis() - Timestamp.currentBucketStartMillis( bucketsPerHour );
      if( elapsed < safeInterval )
         log.debug( "not safe to process yet ({}ms), some of the files could still be open, waiting...", elapsed );
      else for( Path path : Files.wildcard( sourceDirectory, mask ) ) {
         if( path.getFileName().toString().contains( timestamp ) ) {
            log.debug( "skipping (current timestamp) {}", path );
            continue;
         }

         Path targetFile = destinationDirectory.resolve( sourceDirectory.relativize( path )
            + encoding.extension.map( e -> "." + e ).orElse( "" ) );

         Metrics.measureTimer( Metrics.name( "archive" ), () -> {
            if( encoding.compress ) {
               log.debug( "compressing {} ({} bytes)", path, path.toFile().length() );
               Path targetTemp = destinationDirectory.resolve( sourceDirectory.relativize( path ) + ".tmp" );
               Files.copy( path, PLAIN, targetTemp, encoding, bufferSize );

               final Set<String> hourBased = hourBasedFiles.stream()
                  .filter( hourBasedFile -> path.getFileName().toString().contains( hourBasedFile ) )
                  .collect( Collectors.toSet() );

               hourBased.forEach( file ->  {
                  String hourReplacement = StringUtils.replacePattern( Timestamp.parseTimestamp(
                     path.getFileName().toString() ), "-\\d{2}$", "-00" );

                  Path targetHourly = hourBasedDestination.resolve( sourceDirectory.relativize( path ).toString().
                     replace( Timestamp.parseTimestamp( path.getFileName().toString()), hourReplacement ) +
                     encoding.extension.map( e -> "." + e ).orElse( "" ));

                  Files.append( path, PLAIN, targetHourly, encoding, bufferSize );
               });

               Files.rename( targetTemp, targetFile );
               log.debug( "compressed {} ({} bytes)", path, targetFile.toFile().length() );
               Files.delete( path );
            } else {
               log.debug( "moving {} ({} bytes)", path, path.toFile().length() );
               Files.ensureFile( targetFile );
               Files.rename( path, targetFile );
            }
         } );
      }
      log.debug( "packing is done" );
   }
}

@@ -1290,53 +1290,8 @@
 cs;%0A
-import org.apache.commons.lang3.StringUtils;%0A
 impo
@@ -1384,118 +1384,8 @@
 ath;
-%0Aimport java.util.HashSet;%0Aimport java.util.Objects;%0Aimport java.util.Set;%0Aimport java.util.stream.Collectors;
 %0A%0Aim
@@ -1717,103 +1717,8 @@
 ur;%0A
-   private Set%3CString%3E hourBasedFiles = new HashSet%3C%3E();%0A   private Path hourBasedDestination;%0A
    p
@@ -1935,83 +1935,9 @@
 Hour
-, Set%3CString%3E hourBasedFiles,%0A                    Path hourBasedDestination
+ 
 ) %7B%0A
@@ -2182,277 +2182,8 @@
 ur;%0A
-      this.hourBasedFiles = hourBasedFiles;%0A%0A      if (!hourBasedFiles.isEmpty()) %7B%0A         this.hourBasedDestination = Objects.requireNonNull(hourBasedDestination, %22hour based destination has to be %22 +%0A            %22specified, if there are files configured%22);%0A      %7D%0A
    %7D
@@ -3508,851 +3508,8 @@
  );%0A
-%0A               final Set%3CString%3E hourBased = hourBasedFiles.stream()%0A                  .filter( hourBasedFile -%3E path.getFileName().toString().contains( hourBasedFile ) )%0A                  .collect( Collectors.toSet() );%0A%0A               hourBased.forEach( file -%3E  %7B%0A                  String hourReplacement = StringUtils.replacePattern( Timestamp.parseTimestamp(%0A                     path.getFileName().toString() ), %22-%5C%5Cd%7B2%7D$%22, %22-00%22 );%0A%0A                  Path targetHourly = hourBasedDestination.resolve( sourceDirectory.relativize( path ).toString().%0A                     replace( Timestamp.parseTimestamp( path.getFileName().toString()), hourReplacement ) +%0A                     encoding.extension.map( e -%3E %22.%22 + e ).orElse( %22%22 ));%0A%0A                  Files.append( path, PLAIN, targetHourly, encoding, bufferSize );%0A               %7D);%0A%0A
     

Remove hour-based archiving