refactor messages

src/fetch/GET.ts

@@ -1,4 +1,3 @@
-import { checkHealth } from "../health.js";
 import * as prometheus from "../prometheus.js";
 import * as sqlite from "../sqlite.js";
 import { db } from "../../index.js";
@@ -8,8 +7,9 @@ import openapi from "./openapi.js";
 import swaggerHtml  from "../../swagger/index.html"
 import swaggerFavicon from "../../swagger/favicon.png"
 import { toFile, toJSON, toText } from "./cors.js";
-import { recentMessagesEndpoint } from "../recentMessages.js";
+import { selectMessages } from "./messages.js";
 import { DEFAULT_RECENT_MESSAGES_LIMIT } from "../config.js";
+import { checkHealth } from "./health.js";
 
 export default async function (req: Request, server: Server) {
     const { pathname, searchParams} = new URL(req.url);
@@ -24,12 +24,10 @@ export default async function (req: Request, server: Server) {
     let limit = Number(searchParams.get("limit"));
     let sort = searchParams.get("sort");
 
-    //error handling
-
+    // error handling
     if (limit === null || limit === 0) limit = DEFAULT_RECENT_MESSAGES_LIMIT;
     if (isNaN(Number(limit))) return toText("limit must be a number", 400 );
     if (sort === null) sort = "desc";
-    console.log(distinct)
     if (distinct !== "true" && distinct !== null) return toText("distinct must be set to true if declared", 400 );
     if (distinct === "true" && chain) return toText("chain cannot be set if distinct is set to true", 400 );
     if (sort !== "asc" && sort !== "desc") return toText("sort must be asc or desc", 400 );
@@ -48,7 +46,7 @@ export default async function (req: Request, server: Server) {
     if ( pathname === "/traceId") return toJSON(sqlite.selectAll(db, "traceId"));
     if ( pathname === "/chain") return toJSON(sqlite.selectAll(db, "chain"));
     if ( pathname === "/openapi") return toJSON(openapi);
-    if ( pathname === "/recentMessages") return recentMessagesEndpoint(db, chain, moduleHash, limit, distinct, sort);
+    if ( pathname === "/messages") return selectMessages(db, chain, moduleHash, limit, distinct, sort);
 
     return toText("Not found", 400 );
 }

src/fetch/POST.ts

@@ -6,7 +6,7 @@ import { verify } from "../verify.js";
 import { PUBLIC_KEY } from "../config.js";
 import { Server } from "bun";
 import { toText } from "./cors.js";
-import { recentMessages } from "../recentMessages.js";
+import { insertMessages } from "./messages.js";
 
 export default async function (req: Request, server: Server) {
     // get headers and body from POST request
@@ -87,7 +87,7 @@ export default async function (req: Request, server: Server) {
     sqlite.replace(db, "traceId", `${chain}:${traceId}`, timestamp);
     //Set timestamp as key to filter recent messages
 
-    recentMessages( db, traceId, JSON.stringify(json), chain );
+    insertMessages( db, traceId, JSON.stringify(json), chain );
 
     return toText("OK");
 }

src/health.ts → src/fetch/health.ts

@@ -1,4 +1,4 @@
-import { toText } from "./fetch/cors.js";
+import { toText } from "./cors.js";
 
 // https://github.com/pinax-network/substreams-sink-websockets/issues/2#issuecomment-1746121519
 export function checkHealth() {

src/fetch/messages.ts

@@ -0,0 +1,61 @@
+import * as sqlite from "../sqlite.js";
+import { RECENT_MESSAGES_LIMIT } from "../config.js";
+import { toJSON } from "../http.js";
+import Database from "bun:sqlite";
+
+export function insertMessages(db: Database, traceId: string, timestamp: string, chain?: string) {
+    const dbLength = sqlite.count(db, "messages");
+
+    if (dbLength >= RECENT_MESSAGES_LIMIT) {
+        let oldest = sqlite.selectAll(db, "messages").sort((a: any, b: any) => a.timestamp - b.timestamp)[0];
+        console.log("oldest", oldest)
+
+        // update messages
+        sqlite.replaceRecent(db, "messages", String(Date.now()), `${traceId}`, timestamp);
+        sqlite.deleteRow(db, "messages", oldest.key);
+
+        // update messagesByChain
+        if (chain) {
+            oldest = sqlite.selectAll(db, "messagesByChain").sort((a: any, b: any) => a.timestamp - b.timestamp)[0];
+            console.log(oldest)
+            sqlite.replaceRecent(db, "messagesByChain", String(Date.now()), `${chain}:${traceId}`, timestamp );
+            sqlite.deleteRow(db, "messagesByChain", `${oldest.key}`);
+        }
+        return;
+    }
+    // add messages if tables not full
+    sqlite.replaceRecent(db, "messages", String(Date.now()), `${traceId}`, timestamp);
+
+    if (chain) sqlite.replaceRecent(db, "messagesByChain", String(Date.now()), `${chain}:${traceId}`, timestamp );
+}
+
+export function selectMessages(db: Database, chain?: string, moduleHash?: string, limit?: number, distinct?: string, sortBy?: string) {
+
+    let messages = sqlite.selectAllRecent(db,  "messages", "*", sortBy, limit);
+
+    if (distinct) messages = selectDistinct(distinct, messages, db, chain, sortBy, limit);
+    if (chain) messages = sqlite.selectAllRecent(db, "messagesByChain", "*", sortBy, limit).filter((message: any) => message.value.includes(chain));
+    if (moduleHash) messages = messages.filter((message: any) => message.value.includes(moduleHash));
+
+    return toJSON(messages);
+}
+
+export function selectDistinct(distinct?: string, messages?: any, db?: any, chain?: string, sortBy?: string, limit?: number) {
+    let chainList = sqlite.selectAll(db, "chain");
+    if (distinct === "true") {
+
+    let distinctChain = [];
+        for (let i = 0; i < chainList.length; i++) {
+            let chainName = chainList[i].key;
+
+            messages = sqlite.selectAllRecent(db, "messagesByChain", "*", sortBy, limit)
+            messages = messages.filter((message: any) => message.value.includes(chainName));
+
+            distinctChain.push(messages[0]);
+            chainList.slice(i, 1);
+        }
+        let result = distinctChain.filter((notNull) => notNull !== undefined)
+        return result;
+    }
+    return;
+}