add RSS feed endpoint for querying AQ message views

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();
+    }
+}