Added health endpoint and fixed headers

package.json

@@ -1,6 +1,6 @@
 {
     "name": "substreams-sink",
-    "version": "0.9.4",
+    "version": "0.9.5",
     "description": "Substreams Sink",
     "type": "module",
     "exports": "./dist/index.js",

src/config.ts

@@ -11,5 +11,5 @@ export const DEFAULT_COLLECT_DEFAULT_METRICS = false;
 export const DEFAULT_START_BLOCK = "-1";
 export const DEFAULT_SUBSTREAMS_API_TOKEN = "";
 export const DEFAULT_PARAMS = [];
-export const DEFAULT_HEADERS = [];
+export const DEFAULT_HEADERS = new Headers();
 export const DEFAULT_AUTH_ISSUE_URL = "https://auth.pinax.network/v1/auth/issue";

src/health.ts

@@ -0,0 +1,33 @@
+import type { IncomingMessage, ServerResponse } from "http";
+
+import * as prometheus from "./prometheus.js";
+import * as http from "./http.js";
+
+export function health() {
+    http.server.on("request", async (req, res) => {
+        if (!req.url) return;
+        try {
+            if (req.method == "GET") {
+                if (req.url === "/health") {
+                    const messages = await getSingleMetric("substreams_sink_data_message")
+                    if (messages) return toText(res, "OK");
+                    return toText(res, "no messages received yet", 503);
+                }
+            }
+        } catch (err: any) {
+            res.statusCode = 400;
+            return res.end(err.message);
+        }
+    });
+}
+
+async function getSingleMetric(name: string) {
+    const metric = prometheus.registry.getSingleMetric(name);
+    const get = await metric?.get();
+    return get?.values[0].value;
+}
+
+function toText(res: ServerResponse<IncomingMessage>, data: any, status = 200) {
+    res.writeHead(status, { 'Content-Type': 'text/plain; charset=utf-8' });
+    return res.end(data);
+}

src/setup.ts

@@ -10,6 +10,7 @@ import * as prometheus from "./prometheus.js";
 import { logger } from "./logger.js";
 import { onRestartInactivitySeconds } from "./restartInactivitySeconds.js";
 import { applyParams } from "./applyParams.js";
+import { health } from "./health.js";
 
 export async function setup(options: RunOptions) {
     // Configure logging with TSLog
@@ -37,6 +38,9 @@ export async function setup(options: RunOptions) {
     // Adding default headers
     headers.set("User-Agent", "substreams-sink");
 
+    // Health check
+    health();
+
     // Apply params
     if (params.length && substreamPackage.modules) {
         applyParams(params, substreamPackage.modules.modules);