add RSS feed endpoint for querying AQ message views (#1512)

Alternate considerations:

- Looked into using a third-party RSS library for building up the XML.
Seemed like the only somewhat maintained library is Rome, but that
hasn't been updated since 2023 and contains the jdom2 dependency which
has a known vulnerability (wouldn't really matter to our use-case of
producing XML only, but would matter on the client). didn't seem too
much a lift to just use our own DTO's and jackson serialization. If this
ends up being a burden, it also wouldn't be too hard to update to
something else in the future.
- opted to not set the atom links for self or previous as those would be
moving targets given the week lookback limit. We do need a next cursor
so the fact that the URL would not be reproducible after a time period
is unavoidable. We could also throw out the cursor and only use the
`since` field.
- don't search for messages older than a week. This could be shortened
as needed.

Author: adamkorynta

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

@@ -133,6 +133,7 @@
 import cwms.cda.api.rating.RatingTemplateController;
 import cwms.cda.api.rating.ReverseRateTimeSeriesController;
 import cwms.cda.api.rating.ReverseRateValuesController;
+import cwms.cda.api.rss.RssHandler;
 import cwms.cda.api.timeseriesprofile.TimeSeriesProfileCatalogController;
 import cwms.cda.api.timeseriesprofile.TimeSeriesProfileController;
 import cwms.cda.api.timeseriesprofile.TimeSeriesProfileCreateController;
@@ -254,7 +255,8 @@
     "/user/*",
     "/users/*",
     "/roles/*",
-    "/version/*"
+    "/version/*",
+    "/rss/*"
 })
 public class ApiServlet extends HttpServlet {
 
@@ -599,6 +601,7 @@ protected void configureRoutes() {
         addUserManagementHandlers();
 
         get("/version/", new CdaVersionHandler(metrics), requiredRoles);
+        get(format("/rss/{%s}/{%s}", Controllers.OFFICE, Controllers.NAME), new RssHandler(metrics), requiredRoles);
     }
 
     private void addUserManagementHandlers() {

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

@@ -96,6 +96,7 @@ public final class Controllers {
     public static final String NAME = "name";
     public static final String CASCADE_DELETE = "cascade-delete";
     public static final String DATUM = "datum";
+    public static final String SINCE = "since";
     public static final String BEGIN = "begin";
     public static final String END = "end";
     public static final String TIMEZONE = "timezone";

cwms-data-api/src/main/java/cwms/cda/api/rss/RssHandler.java

@@ -0,0 +1,151 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2025 Hydrologic Engineering Center
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package cwms.cda.api.rss;
+
+import static cwms.cda.api.Controllers.CURSOR;
+import static cwms.cda.api.Controllers.GET_ALL;
+import static cwms.cda.api.Controllers.NAME;
+import static cwms.cda.api.Controllers.OFFICE;
+import static cwms.cda.api.Controllers.PAGE;
+import static cwms.cda.api.Controllers.PAGE_SIZE;
+import static cwms.cda.api.Controllers.SINCE;
+import static cwms.cda.api.Controllers.STATUS_200;
+import static cwms.cda.api.Controllers.STATUS_400;
+import static cwms.cda.api.Controllers.STATUS_404;
+import static cwms.cda.api.Controllers.queryParamAsClass;
+import static cwms.cda.api.Controllers.queryParamAsInstant;
+import static cwms.cda.data.dao.JooqDao.getDslContext;
+
+import com.codahale.metrics.MetricRegistry;
+import com.codahale.metrics.Timer;
+import cwms.cda.api.BaseHandler;
+import cwms.cda.api.errors.CdaError;
+import cwms.cda.data.dao.rss.MessageDao;
+import cwms.cda.data.dto.CwmsDTOPaginated;
+import cwms.cda.data.dto.rss.RssFeed;
+import cwms.cda.formatters.ContentType;
+import cwms.cda.formatters.Formats;
+import cwms.cda.helpers.ReplaceUtils;
+import io.javalin.core.util.Header;
+import io.javalin.http.Context;
+import io.javalin.http.HttpCode;
+import io.javalin.plugin.openapi.annotations.OpenApi;
+import io.javalin.plugin.openapi.annotations.OpenApiContent;
+import io.javalin.plugin.openapi.annotations.OpenApiParam;
+import io.javalin.plugin.openapi.annotations.OpenApiResponse;
+import java.net.URISyntaxException;
+import java.net.URLDecoder;
+import java.net.URLEncoder;
+import java.nio.charset.StandardCharsets;
+import java.time.Instant;
+import java.util.function.UnaryOperator;
+import org.apache.http.client.utils.URIBuilder;
+import org.jetbrains.annotations.NotNull;
+import org.jooq.DSLContext;
+
+public final class RssHandler extends BaseHandler {
+
+    private static final int DEFAULT_PAGE_SIZE = 500;
+    private static final String TAG = "RSS";
+
+    public RssHandler(MetricRegistry metrics) {
+        super(metrics);
+    }
+
+    @OpenApi(
+        pathParams = {
+            @OpenApiParam(name = OFFICE, required = true, description = "Office id for feed."),
+            @OpenApiParam(name = NAME, required = true, description = "Specifies the name of the feed. " +
+                "eg TS_STORED, STATUS, REALTIME_OPS")
+        },
+        queryParams = {
+            @OpenApiParam(name = SINCE, description = "The start the feed time window. " +
+                "The endpoint will not retrieve more than the last week of messages."),
+            @OpenApiParam(name = PAGE_SIZE, type = Integer.class, description = "The number of feed items to include."),
+            @OpenApiParam(name = PAGE, description = "This end point can return a lot of data, this "
+                + "identifies where in the request you are. This is an opaque"
+                + " value, and can be obtained from the 'next-page' value in "
+                + "the response.")
+        },
+        responses = {
+            @OpenApiResponse(status = STATUS_200, content = {
+                @OpenApiContent(from = RssFeed.class, type = Formats.RSS)
+            }),
+            @OpenApiResponse(status = STATUS_404, description = "Unknown Feed")
+        },
+        description = "Returns RSS feed items limited to the last week.",
+        tags = {TAG}
+    )
+    @Override
+    public void handle(@NotNull Context ctx) throws Exception {
+        try (final Timer.Context ignored = markAndTime(GET_ALL)) {
+            DSLContext dsl = getDslContext(ctx);
+            String office = ctx.pathParam(OFFICE).toUpperCase();
+            String name = ctx.pathParam(NAME);
+            String formatHeader = ctx.header(Header.ACCEPT);
+            ContentType contentType = Formats.parseHeader(formatHeader, RssFeed.class);
+            String cursor = URLDecoder.decode(queryParamAsClass(ctx, new String[]{PAGE, CURSOR}, String.class, ""),
+                StandardCharsets.UTF_8);
+            if (!CwmsDTOPaginated.CURSOR_CHECK.invoke(cursor)) {
+                ctx.json(new CdaError("cursor or page passed in but failed validation"))
+                    .status(HttpCode.BAD_REQUEST);
+                return;
+            }
+            Instant since = queryParamAsInstant(ctx, SINCE);
+            int pageSize = queryParamAsClass(ctx, new String[]{PAGE_SIZE}, Integer.class, DEFAULT_PAGE_SIZE);
+            MessageDao dao = new MessageDao(dsl);
+            RssFeed feed = dao.retrieveFeed(cursor, pageSize, office, name, since, newLinkTemplate(ctx));
+            String result = Formats.format(contentType, feed);
+            ctx.result(result);
+            ctx.contentType(contentType.toString());
+        }
+    }
+
+    private static String getHost(Context ctx) {
+        String scheme = ctx.header("X-Forwarded-Proto");
+        if (scheme == null) {
+            scheme = "https";
+        }
+        String host = ctx.header("X-Forwarded-Host");
+        if (host == null) {
+            host = ctx.host();
+        }
+        String path = ctx.path();
+        return scheme + "://" + host + path;
+    }
+
+    private UnaryOperator<String> newLinkTemplate(Context ctx)
+        throws URISyntaxException {
+        String pageToken = "{page_token}";
+        String url = new URIBuilder(getHost(ctx))
+            .addParameter(PAGE, pageToken)
+            .build()
+            .toString();
+        return new ReplaceUtils.OperatorBuilder()
+            .withTemplate(url)
+            .withOperatorKey(URLEncoder.encode(pageToken, StandardCharsets.UTF_8))
+            .build();
+    }
+}

cwms-data-api/src/main/java/cwms/cda/data/dao/rss/AqTable.java

@@ -0,0 +1,31 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2025 Hydrologic Engineering Center
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package cwms.cda.data.dao.rss;
+
+enum AqTable {
+    TS_STORED,
+    STATUS,
+    REALTIME_OPS
+}

cwms-data-api/src/main/java/cwms/cda/data/dao/rss/MessageDao.java

@@ -0,0 +1,145 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2025 Hydrologic Engineering Center
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package cwms.cda.data.dao.rss;
+
+import static java.util.stream.Collectors.toList;
+import static org.jooq.impl.DSL.currentTimestamp;
+import static org.jooq.impl.DSL.field;
+import static org.jooq.impl.DSL.name;
+import static org.jooq.impl.DSL.partitionBy;
+import static org.jooq.impl.DSL.rowNumber;
+import static org.jooq.impl.DSL.select;
+import static org.jooq.impl.DSL.table;
+
+import cwms.cda.api.errors.NotFoundException;
+import cwms.cda.data.dao.JooqDao;
+import cwms.cda.data.dto.CwmsDTOPaginated;
+import cwms.cda.data.dto.rss.AtomLink;
+import cwms.cda.data.dto.rss.RssChannel;
+import cwms.cda.data.dto.rss.RssFeed;
+import cwms.cda.data.dto.rss.RssItem;
+import java.sql.Timestamp;
+import java.time.Instant;
+import java.time.ZoneOffset;
+import java.time.ZonedDateTime;
+import java.time.temporal.ChronoUnit;
+import java.util.Optional;
+import java.util.function.UnaryOperator;
+import org.jooq.DSLContext;
+import org.jooq.Field;
+import org.jooq.Record;
+import org.jooq.Result;
+import org.jooq.Table;
+
+public final class MessageDao extends JooqDao<RssFeed> {
+
+
+    public MessageDao(DSLContext dsl) {
+        super(dsl);
+    }
+
+    public RssFeed retrieveFeed(String cursor, int pageSize, String office, String name,
+        Instant since, UnaryOperator<String> urlBuilder) {
+        AqTable aqTable = getAqTable(name);
+        String[] cursorSplit = CwmsDTOPaginated.decodeCursor(cursor);
+        int offset = 0;
+        if(cursorSplit.length == 2) {
+            offset = Integer.parseInt(cursorSplit[0]);
+            pageSize = Integer.parseInt(cursorSplit[1]);
+            since = null;
+        }
+        var items = retrieveMessages(offset, pageSize, since, office, aqTable)
+            .map(record -> {
+                Object userData = record.get("USER_DATA");
+                return MessageUtil.extractPayload(userData).map(p -> rssItem(record, p));
+            }).stream()
+            .filter(Optional::isPresent)
+            .map(Optional::get)
+            .collect(toList());
+        AtomLink nextLink = null;
+        if(items.size() == pageSize) {
+            String nextCursor = CwmsDTOPaginated.encodeCursor(items.size() + offset, pageSize);
+            nextLink = new AtomLink("next", urlBuilder.apply(nextCursor));
+        }
+        String description;
+        switch(aqTable) {
+            case TS_STORED:
+                description = " CWMS messages about time series operations, such as data stored and deleted";
+                break;
+            case STATUS:
+                description = " CWMS general system and application status messages";
+                break;
+            case REALTIME_OPS:
+                description = " CWMS application operational messages";
+                break;
+            default:
+                description = null;
+        }
+        RssChannel channel = new RssChannel(name, nextLink, description, items);
+        return new RssFeed(channel);
+    }
+
+    private static AqTable getAqTable(String name) {
+        try {
+            return AqTable.valueOf(name.toUpperCase());
+        } catch (IllegalArgumentException e) {
+            throw new NotFoundException(e);
+        }
+    }
+
+    private static RssItem rssItem(Record record, String p) {
+        ZonedDateTime enqTimestamp = record.get("ENQ_TIMESTAMP", Timestamp.class)
+            .toInstant().atZone(ZoneOffset.UTC);
+        String msgId = record.get("MSG_ID", String.class);
+        return new RssItem(p, enqTimestamp, msgId);
+    }
+
+    private Result<?> retrieveMessages(int offset, int pageSize, Instant since, String office, AqTable name) {
+        Timestamp sinceTimestamp = since == null ? null : Timestamp.from(since);
+        Table<?> t = table(name("CWMS_20", "AQ$" + office + "_" + name.name() + "_TABLE")).as("t");
+
+        Field<String> MSG_ID         = field("MSG_ID", String.class);
+        Field<Timestamp> ENQ_TS      = field("ENQ_TIMESTAMP", Timestamp.class);
+        Field<Object> USER_DATA      = field("USER_DATA", Object.class);
+        Field<Integer> rn = rowNumber()
+                .over(partitionBy(MSG_ID).orderBy(ENQ_TS.desc()))
+                .as("rn");
+        var condition = ENQ_TS.ge(currentTimestamp().minus(7));
+        if (sinceTimestamp != null) {
+            condition = condition.and(ENQ_TS.gt(sinceTimestamp));
+        }
+        var inner = select(MSG_ID, ENQ_TS, USER_DATA, rn)
+                .from(t)
+                .where(condition)
+                .asTable("x");
+        return dsl.select( MSG_ID, ENQ_TS, USER_DATA)
+            .from(inner)
+            .where(field(name("x", "rn"), Integer.class).eq(1))
+            .orderBy(ENQ_TS.desc())
+            .offset(offset)
+            .limit(pageSize)
+            .fetch();
+    }
+}