Paging blobs

cwms-data-api/src/main/java/cwms/cda/api/BlobController.java

@@ -27,7 +27,6 @@
 import io.javalin.plugin.openapi.annotations.OpenApiRequestBody;
 import io.javalin.plugin.openapi.annotations.OpenApiResponse;
 import java.io.InputStream;
-import java.util.List;
 import java.util.Optional;
 
 import javax.servlet.http.HttpServletResponse;
@@ -117,9 +116,8 @@ public void getAll(@NotNull Context ctx) {
             ContentType contentType = Formats.parseHeader(formatHeader, Blobs.class);
 
             BlobDao dao = new BlobDao(dsl);
-            List<Blob> blobList = dao.getAll(office, like);
+            Blobs blobs = dao.getBlobs(cursor, pageSize, office, like);
 
-            Blobs blobs = new Blobs.Builder(cursor, pageSize, 0).addAll(blobList).build();
             String result = Formats.format(contentType, blobs);
 
             ctx.result(result);

cwms-data-api/src/main/java/cwms/cda/data/dao/BlobDao.java

@@ -2,11 +2,21 @@
 
 import cwms.cda.api.errors.NotFoundException;
 import cwms.cda.data.dto.Blob;
+import cwms.cda.data.dto.Blobs;
+import cwms.cda.data.dto.CwmsDTOPaginated;
 import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+import org.jooq.Condition;
 import org.jooq.DSLContext;
+import org.jooq.Field;
 import org.jooq.Record;
+import org.jooq.Record4;
 import org.jooq.ResultQuery;
+import org.jooq.SelectLimitPercentStep;
+import org.jooq.Table;
 import usace.cwms.db.jooq.codegen.packages.CWMS_TEXT_PACKAGE;
+import usace.cwms.db.jooq.codegen.tables.AV_CWMS_MEDIA_TYPE;
+import usace.cwms.db.jooq.codegen.tables.AV_OFFICE;
 
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
@@ -16,9 +26,24 @@
 import java.sql.SQLException;
 import java.util.List;
 import java.util.Optional;
+import java.util.stream.Stream;
+
+import static org.jooq.impl.DSL.field;
+import static org.jooq.impl.DSL.name;
+import static org.jooq.impl.DSL.noCondition;
+import static org.jooq.impl.DSL.table;
+import static org.jooq.impl.DSL.upper;
 
 public class BlobDao extends JooqDao<Blob> {
 
+    public static final String ID = "ID";
+    public static final String DESCRIPTION = "DESCRIPTION";
+    public static final String OFFICE_CODE = "OFFICE_CODE";
+    public static final String MEDIA_TYPE_CODE = "MEDIA_TYPE_CODE";
+    public static final String MEDIA_TYPE_ID = "MEDIA_TYPE_ID";
+    public static final String VALUE = "VALUE";
+    public static final String OFFICE_ID = "OFFICE_ID";
+
     public static final String BLOB_WITH_OFFICE = "SELECT CWMS_MEDIA_TYPE.MEDIA_TYPE_ID, AT_BLOB.VALUE \n"
             + "FROM CWMS_20.AT_BLOB \n"
             + "join CWMS_20.CWMS_MEDIA_TYPE on AT_BLOB.MEDIA_TYPE_CODE = CWMS_MEDIA_TYPE.MEDIA_TYPE_CODE \n"
@@ -49,11 +74,11 @@ public Optional<Blob> getByUniqueName(String id, String limitToOffice) {
         }
 
         Blob retVal = query.fetchOne(r -> {
-            String rId = r.get("ID", String.class);
-            String rOffice = r.get("OFFICE_ID", String.class);
-            String rDesc = r.get("DESCRIPTION", String.class);
-            String rMedia = r.get("MEDIA_TYPE_ID", String.class);
-            byte[] value = r.get("VALUE", byte[].class);
+            String rId = r.get(ID, String.class);
+            String rOffice = r.get(OFFICE_ID, String.class);
+            String rDesc = r.get(DESCRIPTION, String.class);
+            String rMedia = r.get(MEDIA_TYPE_ID, String.class);
+            byte[] value = r.get(VALUE, byte[].class);
             return new Blob(rOffice, rId, rDesc, rMedia, value);
         });
 
@@ -136,15 +161,94 @@ public List<Blob> getAll(String officeId, String like) {
         }
 
         return query.fetch(r -> {
-            String rId = r.get("ID", String.class);
-            String rOffice = r.get("OFFICE_ID", String.class);
-            String rDesc = r.get("DESCRIPTION", String.class);
-            String rMedia = r.get("MEDIA_TYPE_ID", String.class);
+            String rId = r.get(ID, String.class);
+            String rOffice = r.get(OFFICE_ID, String.class);
+            String rDesc = r.get(DESCRIPTION, String.class);
+            String rMedia = r.get(MEDIA_TYPE_ID, String.class);
 
             return new Blob(rOffice, rId, rDesc, rMedia, null);
         });
     }
 
+    /**
+     * Retrieves all blobs with pagination support.
+     *
+     * @param cursor   the pagination cursor, can be null or empty for the first page
+     * @param pageSize the number of blobs to retrieve per page
+     * @param officeId filter by office ID, can be null or empty to include all offices
+     * @param like     filter blobs by a case-insensitive regex pattern on their IDs, can be null or empty
+     * @return a Blobs object containing the retrieved blobs and pagination information
+     */
+    public @NotNull Blobs getBlobs(@Nullable String cursor, int pageSize, @Nullable String officeId, @Nullable String like) {
+
+        String cursorOffice = null;
+        String cursorId = null;
+
+        AV_CWMS_MEDIA_TYPE cwmsMediaType = AV_CWMS_MEDIA_TYPE.AV_CWMS_MEDIA_TYPE.as("cmt");
+        AV_OFFICE vOffice = AV_OFFICE.AV_OFFICE.as("vo");
+
+        // 2025-07-28 AT_BLOB does not seem to be in the codegen but I'd still like to use the DSL style.
+        // Manually create the blob table and fields.
+        Table<?> atBlob = table(name("CWMS_20", "AT_BLOB")).as("bt");
+        Field<String> blobIdFld = field(name(atBlob.getName(), ID), String.class);
+        Field<String> descFld = field(name(atBlob.getName(), DESCRIPTION), String.class);
+        Field<Long> officeCodeFld = field(name(atBlob.getName(), OFFICE_CODE), Long.class);
+        Field<Long> mediaCodeFld = field(name(atBlob.getName(), MEDIA_TYPE_CODE), Long.class);
+
+        Condition pagingCondition = noCondition();
+        if (cursor != null && !cursor.isEmpty()) {
+            final String[] parts = CwmsDTOPaginated.decodeCursor(cursor, "||");
+
+            if (parts.length > 1) {
+                cursorOffice = Blobs.getOffice(cursor);
+                cursorId = Blobs.getId(cursor);
+
+                pageSize = Integer.parseInt(parts[2]);
+            }
+
+            Condition moreInSameOffice = cursorId == null || cursorOffice == null ? noCondition() :
+                    vOffice.OFFICE_ID.eq(cursorOffice.toUpperCase())
+                            .and(upper(blobIdFld).greaterThan(cursorId.toUpperCase()));
+            Condition nextOffices = cursorOffice == null ? noCondition():
+                    upper(vOffice.OFFICE_ID).greaterThan(cursorOffice.toUpperCase());
+            pagingCondition = moreInSameOffice.or(nextOffices);
+        }
+
+        Condition whereCondition = noCondition();
+
+        if (like != null && !like.isEmpty()) {
+            whereCondition = whereCondition.and(caseInsensitiveLikeRegex(blobIdFld, like));
+        }
+        if(officeId != null && !officeId.isEmpty()) {
+            whereCondition = whereCondition.and(upper(vOffice.OFFICE_ID).eq(upper(officeId)));
+        }
+
+        SelectLimitPercentStep<Record4<String, String, String, String>> query = dsl.select(blobIdFld, descFld, cwmsMediaType.MEDIA_TYPE_ID, vOffice.OFFICE_ID)
+                .from(atBlob)
+                .join(cwmsMediaType).on(mediaCodeFld.eq(cwmsMediaType.MEDIA_TYPE_CODE.cast(Long.class)))
+                .join(vOffice).on(vOffice.OFFICE_CODE.eq(officeCodeFld))
+                .where(whereCondition)
+                .and(pagingCondition)
+                .orderBy(vOffice.OFFICE_ID, blobIdFld)
+                .limit(pageSize);
+
+        Blobs.Builder builder = new Blobs.Builder(cursor, pageSize, 0);
+
+        try (Stream<Record4<String, String, String, String>> stream = query.stream()){
+            stream.forEach(r -> {
+                String rId = r.value1();
+                String rDesc = r.value2();
+                String rMedia = r.value3();
+                String rOffice = r.value4();
+
+                Blob blob = new Blob(rOffice, rId, rDesc, rMedia, null);
+                builder.addBlob(blob);
+            });
+        }
+
+        return builder.build();
+    }
+
     public void create(Blob blob, boolean failIfExists, boolean ignoreNulls) {
         String pFailIfExists = formatBool(failIfExists);
         String pIgnoreNulls = formatBool(ignoreNulls);

cwms-data-api/src/main/java/cwms/cda/data/dto/Blobs.java

@@ -34,6 +34,40 @@ public List<Blob> getBlobs() {
         return Collections.unmodifiableList(blobs);
     }
 
+    /**
+     * Extract the office from the cursor.
+     *
+     * @param cursor the cursor
+     * @return office
+     */
+    public static String getOffice(String cursor) {
+        String[] parts = CwmsDTOPaginated.decodeCursor(cursor);
+        if (parts.length > 1) {
+            String[] officeAndId = CwmsDTOPaginated.decodeCursor(parts[0]);
+            if (officeAndId.length > 0) {
+                return officeAndId[0];
+            }
+        }
+        return null;
+    }
+
+    /**
+     * Extract the id from the cursor.
+     *
+     * @param cursor the cursor
+     * @return id
+     */
+    public static String getId(String cursor) {
+        String[] parts = CwmsDTOPaginated.decodeCursor(cursor);
+        if (parts.length > 1) {
+            String[] officeAndId = CwmsDTOPaginated.decodeCursor(parts[0]);
+            if (officeAndId.length > 1) {
+                return officeAndId[1];
+            }
+        }
+        return null;
+    }
+
     public static class Builder {
         private Blobs workingBlobs;
 
@@ -43,16 +77,17 @@ public Builder(String cursor, int pageSize, int total) {
 
         public Blobs build() {
             if (this.workingBlobs.blobs.size() == this.workingBlobs.pageSize) {
+                Blob lastBlob = this.workingBlobs.blobs.get(this.workingBlobs.blobs.size() - 1);
+                String cursor = encodeCursor(CwmsDTOPaginated.delimiter, lastBlob.getOfficeId(),
+                        lastBlob.getId());
                 this.workingBlobs.nextPage = encodeCursor(
-                        this.workingBlobs.blobs.get(this.workingBlobs.blobs.size() - 1).toString().toUpperCase(),
+                        cursor,
                         this.workingBlobs.pageSize,
                         this.workingBlobs.total);
             } else {
                 this.workingBlobs.nextPage = null;
             }
             return workingBlobs;
-
-
         }
 
         public Builder addBlob(Blob blob) {

cwms-data-api/src/main/java/cwms/cda/data/dto/Clobs.java

@@ -53,9 +53,9 @@ private void addClob(Clob clob) {
     public static String getOffice(String cursor) {
         String[] parts = CwmsDTOPaginated.decodeCursor(cursor);
         if (parts.length > 1) {
-            String[] idAndOffice = CwmsDTOPaginated.decodeCursor(parts[0]);
-            if (idAndOffice.length > 0) {
-                return idAndOffice[0];
+            String[] officeAndId = CwmsDTOPaginated.decodeCursor(parts[0]);
+            if (officeAndId.length > 0) {
+                return officeAndId[0];
             }
         }
         return null;
@@ -70,9 +70,9 @@ public static String getOffice(String cursor) {
     public static String getId(String cursor) {
         String[] parts = CwmsDTOPaginated.decodeCursor(cursor);
         if (parts.length > 1) {
-            String[] idAndOffice = CwmsDTOPaginated.decodeCursor(parts[0]);
-            if (idAndOffice.length > 1) {
-                return idAndOffice[1];
+            String[] officeAndId = CwmsDTOPaginated.decodeCursor(parts[0]);
+            if (officeAndId.length > 1) {
+                return officeAndId[1];
             }
         }
         return null;

cwms-data-api/src/test/java/cwms/cda/api/BlobControllerTestIT.java

@@ -1,11 +1,13 @@
 package cwms.cda.api;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import cwms.cda.data.dto.Blob;
 import cwms.cda.formatters.Formats;
 import cwms.cda.formatters.json.JsonV2;
 import fixtures.TestAccounts;
 import io.restassured.filter.log.LogDetail;
+import io.restassured.response.Response;
 import org.apache.commons.lang3.RandomStringUtils;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Tag;
@@ -16,10 +18,15 @@
 import javax.servlet.http.HttpServletResponse;
 import java.io.UnsupportedEncodingException;
 import java.net.URLEncoder;
+import java.time.Duration;
 
 import static io.restassured.RestAssured.given;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.is;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTimeout;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 @Tag("integration")
 public class BlobControllerTestIT extends DataApiTestIT {
@@ -352,6 +359,93 @@ void test_create_too_long_name() throws Exception
             .body("message", containsString("One or more provided values exceeds the maximum length for the parameter."));
     }
 
+
+    @Test
+    void test_pagination_works() {
+        int count = 11;
+        String prefix = "test_blob_pagination_";
+        assertTimeout(Duration.ofMinutes(5), () -> {
+
+            createBlobs(count, prefix);
+
+            final int pageSize = 5;
+            Response initialResponse =
+                    given()
+                            .log()
+                            .ifValidationFails(LogDetail.ALL, true)
+                            .queryParam(Controllers.OFFICE, SPK)
+                            .queryParam(Controllers.PAGE_SIZE, pageSize)
+                            .queryParam(Controllers.LIKE, prefix + ".+")
+                            .when()
+                            .get("/blobs/" )
+                            .then()
+                            .log().ifValidationFails(LogDetail.ALL, true)
+                            .assertThat()
+                            .statusCode(is(HttpServletResponse.SC_OK))
+                            .extract().response();
+
+            String nextPage = initialResponse.path("next-page");
+            assertNotNull(nextPage, "Expected a next page to be returned");
+            int totalRetrieved = initialResponse.path("blobs.size()");
+            assertEquals(pageSize, totalRetrieved, "Expected the first page to return the configured page size");
+
+            do {
+                Response pageN =
+                        given()
+                                .log()
+                                .ifValidationFails(LogDetail.ALL, true)
+                                .queryParam(Controllers.OFFICE, SPK)
+                                .queryParam("page", nextPage)
+                                .when()
+                                .get("/blobs/" )
+                                .then()
+                                .log().ifValidationFails(LogDetail.ALL, true)
+                                .assertThat()
+                                .statusCode(is(HttpServletResponse.SC_OK))
+                                .extract().response();
+
+                nextPage = pageN.path("next-page");
+                int pageTotal = pageN.path("blobs.size()");
+                assertTrue(pageTotal <= pageSize, "Expected the page to return no more than the configured page size");
+
+                totalRetrieved += pageTotal;
+            } while( nextPage != null );
+
+            assertTrue(totalRetrieved > pageSize, "Expected more than one page of blobs");
+            assertEquals(count, totalRetrieved, "Expected to retrieve the blobs we created. ");
+
+
+        }, "Blobs retrieval got stuck; possibly in endless loop");
+    }
+
+    private static void createBlobs(int count, String prefix) throws JsonProcessingException {
+        //
+        TestAccounts.KeyUser user = TestAccounts.KeyUser.SPK_NORMAL;
+        String mediaType = "application/octet-stream";
+        ObjectMapper om = JsonV2.buildObjectMapper();
+        for( int i = 0; i < count; i++ ) {
+            String testValue = "test value " + i ;
+            Blob blob = new Blob(SPK, prefix + i, "test blob number "+i, mediaType, testValue.getBytes() );
+            String serializedBlob = om.writeValueAsString(blob);
+            given()
+                    .log().ifValidationFails(LogDetail.ALL,true)
+                    .contentType(Formats.JSONV2)
+                    .body(serializedBlob)
+                    .header("Authorization",user.toHeaderValue())
+                    .queryParam("office",SPK)
+                    .queryParam("fail-if-exists",false)
+                    .when()
+                    .redirects().follow(true)
+                    .redirects().max(3)
+                    .post("/blobs/")
+                    .then()
+                    .log().ifValidationFails(LogDetail.ALL,true)
+                    .assertThat()
+                    .statusCode(is(HttpServletResponse.SC_CREATED));
+        }
+    }
+
+
     enum GetAllTest
     {
         DEFAULT(Formats.DEFAULT, Formats.JSONV2),