fix(stac-loader): only ingest the last item/collection for a given id in a batch (#227)

* fix(stac-loader): only ingest the last item/collection for a given id in a batch

* fix(stac-loader): reduce number of stac-loader redrives from 5 to 2

* chore(stac-loader): refactor stac-loader handler into component functions

Author: hrodmn

lib/stac-loader/index.ts

@@ -195,6 +195,29 @@ export interface StacLoaderProps {
  * This approach balances throughput (larger batches = fewer database connections)
  * with latency (time-based triggers prevent indefinite waiting).
  *
+ * ## Message Ordering and Deduplication
+ *
+ * **Standard Queues**: This construct uses standard (non-FIFO) SNS topics and SQS queues,
+ * which means messages are **not guaranteed to arrive in order**. Multiple messages
+ * with the same STAC item or collection ID may arrive in any sequence.
+ *
+ * **Timestamp-Based Deduplication**: Within each batch, the loader uses SNS timestamps
+ * to ensure only the newest version of each item/collection is ingested:
+ *
+ * - When multiple messages have the same item/collection ID in a batch, the loader
+ *   compares their SNS Timestamps (automatically set when messages are published)
+ * - Only the message with the **newest timestamp** is kept for database insertion
+ * - Older versions are discarded and logged at the debug level
+ * - This guarantees that within a batch, the chronologically latest update wins
+ *
+ * **Important Limitations**:
+ * - Deduplication only occurs **within a single batch** - messages in different batches
+ *   are not compared across batches
+ * - The database upsert operation will update existing records, so later batches can
+ *   still overwrite earlier batches regardless of timestamps
+ * - For guaranteed ordering across all messages, consider implementing version tracking
+ *   in your STAC metadata and database constraints
+ *
  * ## Error Handling and Dead Letter Queue
  *
  * Failed messages are sent to the dead letter queue after 5 processing attempts.
@@ -396,7 +419,7 @@ export class StacLoader extends Construct {
       visibilityTimeout: Duration.seconds(timeoutSeconds + 10),
       encryption: sqs.QueueEncryption.SQS_MANAGED,
       deadLetterQueue: {
-        maxReceiveCount: 5,
+        maxReceiveCount: 2,
         queue: this.deadLetterQueue,
       },
     });
@@ -408,7 +431,7 @@ export class StacLoader extends Construct {
 
     // Subscribe the queue to the topic
     this.topic.addSubscription(
-      new snsSubscriptions.SqsSubscription(this.queue)
+      new snsSubscriptions.SqsSubscription(this.queue),
     );
 
     // Create the lambda function
@@ -448,11 +471,11 @@ export class StacLoader extends Construct {
       new lambdaEventSources.SqsEventSource(this.queue, {
         batchSize: props.batchSize ?? 500,
         maxBatchingWindow: Duration.minutes(
-          props.maxBatchingWindowMinutes ?? 1
+          props.maxBatchingWindowMinutes ?? 1,
         ),
         maxConcurrency: maxConcurrency,
         reportBatchItemFailures: true,
-      })
+      }),
     );
 
     // Create outputs
@@ -490,7 +513,7 @@ export class StacItemLoader extends StacLoader {
   constructor(scope: Construct, id: string, props: StacLoaderProps) {
     console.warn(
       `StacItemLoader is deprecated. Please use StacLoader instead. ` +
-        `StacItemLoader will be removed in a future version.`
+        `StacItemLoader will be removed in a future version.`,
     );
 
     super(scope, id, props);