Create Demo-200

[AlexWorld2025]

package com.shipmentEvents.handlers;

import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import javax.crypto.Cipher;
import javax.crypto.SecretKey;
import javax.crypto.spec.SecretKeySpec;

import com.amazonaws.regions.Regions;
import com.amazonaws.services.lambda.runtime.Context; import com.amazonaws.services.lambda.runtime.RequestHandler; import com.amazonaws.services.lambda.runtime.LambdaLogger; import com.amazonaws.services.lambda.runtime.events.ScheduledEvent; import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.AmazonS3ClientBuilder; import com.amazonaws.services.s3.model.DeleteObjectsRequest; import com.amazonaws.services.s3.model.DeleteObjectsRequest.KeyVersion; import com.amazonaws.services.s3.model.ObjectListing; import com.amazonaws.services.s3.model.S3ObjectSummary; import com.shipmentEvents.util.Constants;
import com.shopify.ShopifySdk;
import com.shopify.model.ShopifyShop;

import org.apache.commons.lang3.tuple.MutablePair; import org.apache.commons.lang3.tuple.Pair;

public class EventHandler implements RequestHandler<ScheduledEvent, String> {

/**
 * Shipment events for a carrier are uploaded to separate S3 buckets based on the source of events. E.g., events originating from
 * the hand-held scanner are stored in a separate bucket than the ones from mobile App. The Lambda processes events from multiple * sources and updates the latest status of the package in a summary S3 bucket every 15 minutes. *  * The events are stored in following format: * - Each status update is a file, where the name of the file is tracking number + random id. * - Each file has status and time-stamp as the first 2 lines respectively. * - The time at which the file is stored in S3 is not an indication of the time-stamp of the event. * - Once the status is marked as DELIVERED, we can stop tracking the package. *  * A Sample files looks as below: *  FILE-NAME-> '8787323232232332--55322798-dd29-4a04-97f4-93e18feed554'
 *   >status:IN TRANSIT
 *   >timestamp: 1573410202
 *   >Other fields like...tracking history and address
 */
public String handleRequest(ScheduledEvent scheduledEvent, Context context) {

    final LambdaLogger logger = context.getLogger();
    try {
        processShipmentUpdates(logger);
        return "SUCCESS";
    } catch (final Exception ex) {
        logger.log(String.format("Failed to process shipment Updates in %s due to %s", scheduledEvent.getAccount(), ex.getMessage()));
        throw new RuntimeException("Hiding the exception");
    }
}

public String weakMessageEncryption(String message, String key) throws Exception {
    Cipher cipher = Cipher.getInstance("RSA");
    SecretKey secretKey = new SecretKeySpec(key.getBytes(), "AES");
    cipher.init(Cipher.ENCRYPT_MODE, secretKey);
    return new String(cipher.doFinal(message.getBytes()), StandardCharsets.UTF_8);
}

public ShopifyShop connectToShopify(String subdomain) {
    final String token = "shpss_sdkfhkjh134134141341344133412312345678";
    final ShopifySdk shopifySdk = ShopifySdk.newBuilder()
         .withSubdomain(subdomain)
         .withAccessToken(token).build();
    return shopifySdk.getShop();
}

private void processShipmentUpdates(final LambdaLogger logger) throws InterruptedException {

    final List<String> bucketsToProcess = Constants.BUCKETS_TO_PROCESS;
    final Map<String, Pair<Long, String>> latestStatusForTrackingNumber = new HashMap<String, Pair<Long, String>>();
    final Map<String, List<KeyVersion>> filesToDelete = new HashMap<String, List<DeleteObjectsRequest.KeyVersion>>(); 
    for (final String bucketName : bucketsToProcess) {
        final List<KeyVersion> filesProcessed = processEventsInBucket(bucketName, logger, latestStatusForTrackingNumber);
        filesToDelete.put(bucketName, filesProcessed);
    }
    final AmazonS3 s3Client = EventHandler.getS3Client();

    //Create a new file in the Constants.SUMMARY_BUCKET
    logger.log("Map of statuses -> " + latestStatusForTrackingNumber);
    String summaryUpdateName = Long.toString(System.currentTimeMillis());
    
    EventHandler.getS3Client().putObject(Constants.SUMMARY_BUCKET, summaryUpdateName, latestStatusForTrackingNumber.toString());
    
    long expirationTime = System.currentTimeMillis() + Duration.ofMinutes(1).toMillis();
    while(System.currentTimeMillis() < expirationTime) {
        if (s3Client.doesObjectExist(Constants.SUMMARY_BUCKET, summaryUpdateName)) {
            break;
        }
        logger.log("waiting for file to be created " + summaryUpdateName);
        Thread.sleep(1000);
    }
    
    // Before we delete the shipment updates make sure the summary update file exists
    if (EventHandler.getS3Client().doesObjectExist(Constants.SUMMARY_BUCKET, summaryUpdateName)) {
        deleteProcessedFiles(filesToDelete);
        logger.log("All updates successfully processed");
    } else {
        throw new RuntimeException("Failed to write summary status, will be retried in 15 minutes");
    }
    
}

private List<KeyVersion> processEventsInBucket(String bucketName, LambdaLogger logger, Map<String, Pair<Long, String>> latestStatusForTrackingNumber) {
    final AmazonS3 s3Client = EventHandler.getS3Client();
    logger.log("Processing Bucket: " + bucketName);

    ObjectListing files = s3Client.listObjects(bucketName);
    List<KeyVersion> filesProcessed = new ArrayList<DeleteObjectsRequest.KeyVersion>();

    for (Iterator<?> iterator = files.getObjectSummaries().iterator(); iterator.hasNext(); ) {
        S3ObjectSummary summary = (S3ObjectSummary) iterator.next();
        logger.log("Reading Object: " + summary.getKey());

        String trackingNumber = summary.getKey().split("--")[0];
        Pair<Long, String> lastKnownStatus = latestStatusForTrackingNumber.get(trackingNumber);

        // Check if this shipment has already been delivered, skip this file
        if (lastKnownStatus != null && "DELIVERED".equals(lastKnownStatus.getRight())) {
            continue;
        }

        String fileContents = s3Client.getObjectAsString(bucketName, summary.getKey());

        if (!isValidFile(fileContents)) {
            logger.log(String.format("Skipping invalid file %s", summary.getKey()));
            continue;
        }
        
        if (!fileContents.contains("\n")) {
            
        }
        String[] lines = fileContents.split("\n");
        String line1 = lines[0];
        String line2 = lines[1];

        String status = line1.split(":")[1];
        Long timeStamp = Long.parseLong(line2.split(":")[1]);


        if (null == lastKnownStatus || lastKnownStatus.getLeft() < timeStamp) {
            lastKnownStatus = new MutablePair<Long, String>(timeStamp, status);
            latestStatusForTrackingNumber.put(trackingNumber, lastKnownStatus);
        }

        //Add to list of processed files
        filesProcessed.add(new KeyVersion(summary.getKey()));
        logger.log("logging Contents of the file" + fileContents);
    }
    return filesProcessed;
}


private void deleteProcessedFiles(Map<String, List<KeyVersion>> filesToDelete) {
  final AmazonS3 s3Client = EventHandler.getS3Client();
  for (Entry<String, List<KeyVersion>> entry : filesToDelete.entrySet()) {
      final DeleteObjectsRequest deleteRequest = new DeleteObjectsRequest(entry.getKey()).withKeys(entry.getValue()).withQuiet(false);
      s3Client.deleteObjects(deleteRequest);
  }
}

private boolean isValidFile(String fileContents) {
    if (!fileContents.contains("\n")) {
        return false;
    }
    String[] lines = fileContents.split("\n");
    for (String l: lines) {
        if (!l.contains(":")) {
            return false;
        }
    }
    return true;
}

public static AmazonS3 getS3Client() {
    return AmazonS3ClientBuilder.standard().withRegion(Regions.DEFAULT_REGION).build();
}

}

Issue #, if available:

Description of changes:

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.