package com.timgroup.eventstore.archiver;

import com.amazonaws.services.s3.AmazonS3;
import com.codahale.metrics.MetricRegistry;
import com.timgroup.eventstore.api.EventSource;
import com.timgroup.eventstore.archiver.monitoring.S3ArchiveBucketConfigurationComponent;
import com.timgroup.eventsubscription.SubscriptionBuilder;
import com.timgroup.remotefilestorage.s3.S3ClientFactory;
import com.timgroup.remotefilestorage.s3.S3DownloadableStorage;
import com.timgroup.remotefilestorage.s3.S3DownloadableStorageWithoutDestinationFile;
import com.timgroup.remotefilestorage.s3.S3ListableStorage;
import com.timgroup.remotefilestorage.s3.S3UploadableStorage;
import com.timgroup.remotefilestorage.s3.S3UploadableStorageForInputStream;
import com.timgroup.tucker.info.Component;

import java.io.IOException;
import java.nio.file.Files;
import java.time.Clock;
import java.util.Collections;
import java.util.List;
import java.util.Properties;

public class S3ArchiverFactory {

    public static final String CONFIG_EVENTSTORE_S3_ARCHIVE_BUCKETNAME = "tg.eventstore.archive.bucketName";
    public static final String CONFIG_EVENTSTORE_S3_ARCHIVE_OBJECT_PREFIX = "tg.eventstore.archive.object.prefix";
    public static final String S3_ARCHIVER_TMP_DIR_PREFIX = "s3archiver-tmp";

    private final String eventStoreId;
    private final String bucketName;
    private final MetricRegistry metricRegistry;
    private final Clock clock;
    private final Properties config;

    private static final int MAX_KEYS_PER_S3_LISTING = 1_000;
    private final AmazonS3 amazonS3;

    /**
     Configures eventStoreId and bucketName from properties set in config.
     @see #CONFIG_EVENTSTORE_S3_ARCHIVE_BUCKETNAME
     @see #CONFIG_EVENTSTORE_S3_ARCHIVE_OBJECT_PREFIX

     @Deprecated use S3ArchiverFactory constructor that does not assign eventStoreId on construction
     */
    @Deprecated
    public S3ArchiverFactory(Properties config, MetricRegistry metricRegistry, Clock clock) {
        this.config = config;
        this.eventStoreId = config.getProperty(CONFIG_EVENTSTORE_S3_ARCHIVE_OBJECT_PREFIX);
        this.bucketName = config.getProperty(CONFIG_EVENTSTORE_S3_ARCHIVE_BUCKETNAME);
        this.metricRegistry = metricRegistry;
        this.clock = clock;
        this.amazonS3 = new S3ClientFactory().fromProperties(config);
    }

    public S3ArchiverFactory(String bucketName, Properties config, MetricRegistry metricRegistry, Clock clock) {
        this.config = config;
        this.eventStoreId = null;
        this.bucketName = bucketName;
        this.metricRegistry = metricRegistry;
        this.clock = clock;
        this.amazonS3 = new S3ClientFactory().fromProperties(config);
    }

    public S3Archiver newS3Archiver(EventSource liveEventSource, int batchsize,  String appName) {
        return build(liveEventSource, new FixedNumberOfEventsBatchingPolicy(batchsize), this.eventStoreId, appName, "Event", S3Archiver.DEFAULT_MONITORING_PREFIX);
    }

    public S3Archiver newS3Archiver(String eventStoreId, EventSource liveEventSource, BatchingPolicy batchingPolicy, String appName) {
        return build(liveEventSource, batchingPolicy, eventStoreId, appName, eventStoreId + "-Archiver", S3Archiver.DEFAULT_MONITORING_PREFIX + "." + eventStoreId);
    }

    private S3Archiver build(EventSource liveEventSource, BatchingPolicy batchingPolicy, String eventStoreId, String appName, String subscriptionName, String monitoringPrefix) {
        List<Component> monitoring = Collections.singletonList(new S3ArchiveBucketConfigurationComponent(amazonS3, bucketName));
        S3UploadableStorageForInputStream s3UploadableStorage = createUploadableStorage();

        return S3Archiver.newS3Archiver(
                liveEventSource,
                s3UploadableStorage,
                eventStoreId,
                SubscriptionBuilder.eventSubscription(subscriptionName),
                batchingPolicy,
                newS3ArchiveMaxPositionFetcher(),
                appName,
                metricRegistry,
                monitoringPrefix,
                clock,
                monitoring);
    }

    public EventSource createS3ArchivedEventSource() {
        return new S3ArchivedEventSource(createS3ListableStorage(amazonS3), createDownloadableStorage(amazonS3), eventStoreId);
    }

    private S3DownloadableStorageWithoutDestinationFile createDownloadableStorage(AmazonS3 amazonS3)  {
        try {
            return new S3DownloadableStorageWithoutDestinationFile(
                    new S3DownloadableStorage(amazonS3, Files.createTempDirectory(S3_ARCHIVER_TMP_DIR_PREFIX), bucketName),
                    amazonS3, bucketName);
        } catch (IOException e) {
            throw new RuntimeException("Failed to create tmp dir for Downloadable storage", e);
        }
    }

    public S3ArchiveMaxPositionFetcher newS3ArchiveMaxPositionFetcher() {
        final S3ListableStorage s3ListableStorage = createS3ListableStorage(amazonS3);
        return new S3ArchiveMaxPositionFetcher(s3ListableStorage, eventStoreId);
    }

    private S3ListableStorage createS3ListableStorage(AmazonS3 amazonS3) {
        return new S3ListableStorage(amazonS3, bucketName, MAX_KEYS_PER_S3_LISTING);
    }

    private S3UploadableStorageForInputStream createUploadableStorage() {
        return new S3UploadableStorageForInputStream(new S3UploadableStorage(amazonS3, bucketName), amazonS3, bucketName);
    }
}

@@ -850,16 +850,45 @@
 .Clock;%0A
+import java.util.Collection;%0A
 import j
@@ -909,16 +909,16 @@
 ctions;%0A
-
 import j
@@ -1591,16 +1591,56 @@
 mazonS3;
+%0A    private List%3CComponent%3E monitoring;
 %0A%0A    /*
@@ -2386,32 +2386,150 @@
 erties(config);%0A
+        this.monitoring = Collections.singletonList(new S3ArchiveBucketConfigurationComponent(amazonS3, bucketName));%0A
     %7D%0A%0A    publi
@@ -2874,24 +2874,142 @@
 es(config);%0A
+        this.monitoring = Collections.singletonList(new S3ArchiveBucketConfigurationComponent(amazonS3, bucketName));%0A
     %7D%0A%0A    p
@@ -3762,137 +3762,8 @@
 ) %7B%0A
-        List%3CComponent%3E monitoring = Collections.singletonList(new S3ArchiveBucketConfigurationComponent(amazonS3, bucketName));%0A
     
@@ -4260,16 +4260,16 @@
  clock,%0A
-
         
@@ -4288,32 +4288,114 @@
 toring);%0A    %7D%0A%0A
+    public Collection%3CComponent%3E monitoring() %7B%0A        return monitoring;%0A    %7D%0A%0A
     public Event

make the S3Archiver bucket-wide monitoring available from the Factory.