package org.zalando.stups.fullstop.jobs.rds;

import com.amazonaws.AmazonServiceException;
import com.amazonaws.regions.Region;
import com.amazonaws.regions.Regions;
import com.amazonaws.services.rds.AmazonRDSClient;
import com.amazonaws.services.rds.model.DBInstance;
import com.amazonaws.services.rds.model.DescribeDBInstancesRequest;
import com.amazonaws.services.rds.model.DescribeDBInstancesResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.zalando.stups.fullstop.aws.ClientProvider;
import org.zalando.stups.fullstop.jobs.FullstopJob;
import org.zalando.stups.fullstop.jobs.common.AccountIdSupplier;
import org.zalando.stups.fullstop.jobs.config.JobsProperties;
import org.zalando.stups.fullstop.violation.Violation;
import org.zalando.stups.fullstop.violation.ViolationBuilder;
import org.zalando.stups.fullstop.violation.ViolationSink;

import javax.annotation.PostConstruct;
import java.util.Map;

import static com.google.common.collect.Maps.newHashMap;
import static org.zalando.stups.fullstop.violation.ViolationType.UNSECURED_PUBLIC_ENDPOINT;

@Component
public class FetchRdsJob implements FullstopJob {


    private static final String EVENT_ID = "checkRdsJob";

    private final Logger log = LoggerFactory.getLogger(FetchRdsJob.class);

    private final AccountIdSupplier allAccountIds;

    private final ClientProvider clientProvider;

    private final JobsProperties jobsProperties;

    private final ViolationSink violationSink;

    @Autowired
    public FetchRdsJob(final AccountIdSupplier allAccountIds, final ClientProvider clientProvider,
                       final JobsProperties jobsProperties,
                       final ViolationSink violationSink) {
        this.allAccountIds = allAccountIds;
        this.clientProvider = clientProvider;
        this.jobsProperties = jobsProperties;
        this.violationSink = violationSink;
    }

    @PostConstruct
    public void init() {
        log.info("{} initialized", getClass().getSimpleName());
    }

    @Scheduled(fixedRate = 300_000)
    public void run() {
        for (final String accountId : allAccountIds.get()) {
            final Map<String, Object> metadata = newHashMap();
            for (final String region : jobsProperties.getWhitelistedRegions()) {
                try {
                    final DescribeDBInstancesResult describeDBInstancesResult = getRds(accountId, region);

                    describeDBInstancesResult.getDBInstances().stream()
                            .filter(DBInstance::getPubliclyAccessible)
                            .filter(dbInstance -> dbInstance.getEndpoint() != null)
                            .forEach(dbInstance -> {
                                metadata.put("unsecuredDatabase", dbInstance.getEndpoint().getAddress());
                                metadata.put("errorMessages", "Unsecured Database! Your DB can be reached from outside");
                                writeViolation(accountId, region, metadata, dbInstance.getEndpoint().getAddress());

                            });

                } catch (final AmazonServiceException a) {

                    if (a.getErrorCode().equals("RequestLimitExceeded")) {
                        log.warn("RequestLimitExceeded for account: {}", accountId);
                    } else {
                        log.error(a.getMessage(), a);
                    }

                }
            }
        }
    }

    private void writeViolation(final String account, final String region, final Object metaInfo, final String rdsEndpoint) {
        final ViolationBuilder violationBuilder = new ViolationBuilder();
        final Violation violation = violationBuilder.withAccountId(account)
                .withRegion(region)
                .withPluginFullyQualifiedClassName(FetchRdsJob.class)
                .withType(UNSECURED_PUBLIC_ENDPOINT)
                .withMetaInfo(metaInfo)
                .withEventId(EVENT_ID)
                .withInstanceId(rdsEndpoint)
                .build();
        violationSink.put(violation);
    }

    private DescribeDBInstancesResult getRds(final String accountId, final String region) {
        final DescribeDBInstancesRequest describeDBInstancesRequest = new DescribeDBInstancesRequest();
        final DescribeDBInstancesResult describeDBInstancesResult;

        final AmazonRDSClient amazonRDSClient = clientProvider.getClient(AmazonRDSClient.class, accountId,
                Region.getRegion(Regions.fromName(region)));
        describeDBInstancesResult = amazonRDSClient.describeDBInstances(describeDBInstancesRequest);


        return describeDBInstancesResult;
    }
}

@@ -1096,16 +1096,43 @@
 til.Map;
+%0Aimport java.util.Optional;
 %0A%0Aimport
@@ -1178,24 +1178,87 @@
 newHashMap;%0A
+import static org.apache.commons.lang3.StringUtils.trimToNull;%0A
 import stati
@@ -2589,90 +2589,428 @@
 nal 
-DescribeDBInstancesResult describeDBInstancesResult = getRds(accountId, region
+AmazonRDSClient amazonRDSClient = clientProvider.getClient(AmazonRDSClient.class, accountId,%0A                            Region.getRegion(Regions.fromName(region)));%0A%0A                    Optional%3CString%3E marker = Optional.empty();%0A%0A                    do %7B%0A                        final DescribeDBInstancesRequest request = new DescribeDBInstancesRequest();%0A                        marker.ifPresent(request::setMarker
 );%0A
-%0A
     
@@ -3025,36 +3025,219 @@
         
-describeDBInstancesR
+    final DescribeDBInstancesResult result = amazonRDSClient.describeDBInstances(request);%0A                        marker = Optional.ofNullable(trimToNull(result.getMarker()));%0A%0A                        r
 esult.ge
@@ -3260,16 +3260,20 @@
 tream()%0A
+    
         
@@ -3355,32 +3355,36 @@
                 
+    
 .filter(dbInstan
@@ -3451,16 +3451,20 @@
         
+ 
+   
 .forEach
@@ -3480,16 +3480,20 @@
 ce -%3E %7B%0A
+    
         
@@ -3614,32 +3614,36 @@
                 
+    
 metadata.put(%22er
@@ -3748,16 +3748,20 @@
         
+  
+  
 writeVio
@@ -3865,16 +3865,20 @@
         
+    
 %7D);%0A%0A   
@@ -3894,238 +3894,99 @@
     
-%7D catch (final AmazonServiceException a) %7B%0A%0A                    if (a.getErrorCode().equals(%22RequestLimitExceeded%22)) %7B%0A                        log.warn(%22RequestLimitExceeded for account: %7B%7D%22, accountId);%0A                    %7D else
+    %7D while (marker.isPresent());%0A%0A                %7D catch (final AmazonServiceException a)
  %7B%0A 
@@ -4000,28 +4000,24 @@
             
- 
-   
 log.error(a.
@@ -4038,31 +4038,8 @@
 a);%0A
-                    %7D%0A%0A
     
@@ -4715,591 +4715,7 @@
    %7D
-%0A%0A    private DescribeDBInstancesResult getRds(final String accountId, final String region) %7B%0A        final DescribeDBInstancesRequest describeDBInstancesRequest = new DescribeDBInstancesRequest();%0A        final DescribeDBInstancesResult describeDBInstancesResult;%0A%0A        final AmazonRDSClient amazonRDSClient = clientProvider.getClient(AmazonRDSClient.class, accountId,%0A                Region.getRegion(Regions.fromName(region)));%0A        describeDBInstancesResult = amazonRDSClient.describeDBInstances(describeDBInstancesRequest);%0A%0A%0A        return describeDBInstancesResult;%0A    %7D
 %0A%7D%0A

handle pagination in FetchRdsJob