From 5cde5b23879dd00ca713335c6efe7fd057f4acb6 Mon Sep 17 00:00:00 2001
From: Grejdi Gjura <ggjura@mbta.com>
Date: Tue, 14 Oct 2025 10:40:47 -0400
Subject: [PATCH] fix: limit objects to insert

---
 .../lib/ex_cubic_ingestion/schema/cubic_load.ex  |  8 +++++---
 .../schema/cubic_load_test.exs                   | 16 ++++++++++++++++
 2 files changed, 21 insertions(+), 3 deletions(-)

diff --git a/ex_cubic_ingestion/lib/ex_cubic_ingestion/schema/cubic_load.ex b/ex_cubic_ingestion/lib/ex_cubic_ingestion/schema/cubic_load.ex
index d344b52..6748508 100644
--- a/ex_cubic_ingestion/lib/ex_cubic_ingestion/schema/cubic_load.ex
+++ b/ex_cubic_ingestion/lib/ex_cubic_ingestion/schema/cubic_load.ex
@@ -97,7 +97,7 @@ defmodule ExCubicIngestion.Schema.CubicLoad do
   From a list of S3 objects defined as maps, filter in only new objects since last load and
   insert them in database.
   """
-  @spec insert_new_from_objects_with_table([map()], CubicTable.t()) ::
+  @spec insert_new_from_objects_with_table([map()], CubicTable.t(), integer()) ::
           {:ok,
            {CubicOdsTableSnapshot.t() | nil,
             [
@@ -105,7 +105,7 @@ defmodule ExCubicIngestion.Schema.CubicLoad do
                CubicOdsTableSnapshot.t() | nil}
             ]}}
           | {:error, term()}
-  def insert_new_from_objects_with_table(objects, table) do
+  def insert_new_from_objects_with_table(objects, table, limit \\ 500) do
     # get ODS snapshot if available
     ods_table_snapshot = CubicOdsTableSnapshot.get_by(table_id: table.id)
 
@@ -122,12 +122,14 @@ defmodule ExCubicIngestion.Schema.CubicLoad do
     # create a list of objects that have a last modified that is later than
     # the last object we have in database
     new_objects =
-      Enum.filter(objects, fn object ->
+      objects
+      |> Enum.filter(fn object ->
         last_modified = parse_and_drop_msec(object.last_modified)
 
         is_nil(last_inserted_load) or
           DateTime.compare(last_modified, last_inserted_load.s3_modified) == :gt
       end)
+      |> Enum.take(limit)
 
     if Enum.empty?(new_objects) do
       {:ok, {ods_table_snapshot, []}}
diff --git a/ex_cubic_ingestion/test/ex_cubic_ingestion/schema/cubic_load_test.exs b/ex_cubic_ingestion/test/ex_cubic_ingestion/schema/cubic_load_test.exs
index d8b2a4f..3253b77 100644
--- a/ex_cubic_ingestion/test/ex_cubic_ingestion/schema/cubic_load_test.exs
+++ b/ex_cubic_ingestion/test/ex_cubic_ingestion/schema/cubic_load_test.exs
@@ -154,6 +154,22 @@ defmodule ExCubicIngestion.Schema.CubicLoadTest do
                ]}} = CubicLoad.insert_new_from_objects_with_table(ods_load_objects, ods_table)
     end
 
+    test "limiting list of ODS objects", %{
+      ods_table: ods_table,
+      ods_load_objects: ods_load_objects
+    } do
+      {:ok, {_last_ods_table_snapshot, new_ods_load_recs}} =
+        CubicLoad.insert_new_from_objects_with_table(ods_load_objects, ods_table, 1)
+
+      assert [List.first(ods_load_objects).key] ==
+               new_ods_load_recs
+               |> Enum.reverse()
+               |> Enum.map(fn {%CubicLoad{status: "ready", s3_key: s3_key}, _ods_load_snapshot,
+                               _table, _ods_table_snapshot} ->
+                 s3_key
+               end)
+    end
+
     test "providing an empty list of objects", %{
       dmap_table: dmap_table
     } do