🥅 improve error (#404)

* 🥅 improve error

* 🥅 improve error

* 🥅 improve error

* 🥅 improve error

* 😭 shoud fix bun issue with request

* chore: use retry policy

* 😭 shoud fix bun issue with request keepalive

* 📝 more logs

* 📝 more logs

* 😭 lowering retry

* 💩 handling poop url

* 🚀

* 🧹

* 🧹

* 🧹

* 🧪

---------

Co-authored-by: mxkaske <[e-mail]>

apps/server/.env.example

···
 UNKEY_TOKEN=
 TINY_BIRD_API_KEY=
 UPSTASH_REDIS_REST_URL=
-UPSTASH_REDIS_REST_TOKEN=
+UPSTASH_REDIS_REST_TOKEN=
+RESEND_API_KEY=

apps/server/fly.toml

···
   processes = ["app"]
 
 [deploy]
-  strategy = "bluegreen"
+  strategy = "rolling"
 
 [[http_service.checks]]
   grace_period = "10s"

apps/server/src/checker/alerting.ts

···
 import { db, eq, schema } from "@openstatus/db";
 import { selectNotificationSchema } from "@openstatus/db/src/schema";
 
-import { monitor } from "./checker";
+import { publishPingRetryPolicy } from "./checker";
 import type { Payload } from "./schema";
 import { providerToFunction } from "./utils";
 
 export async function catchTooManyRetry(payload: Payload) {
-  await monitor({ monitorInfo: payload, latency: -1, statusCode: 500 });
+  await publishPingRetryPolicy({ payload, latency: -1, statusCode: 500 });
   if (payload?.status !== "error") {
     await triggerAlerting({ monitorId: payload.monitorId });
     await updateMonitorStatus({

apps/server/src/checker/checker.test.ts

···
 } from "@openstatus/tinybird";
 
 import * as alerts from "./alerting";
-import { checker } from "./checker";
+import { checkerRetryPolicy } from "./checker";
 
 vi.mock("@openstatus/tinybird", async () => {
   const actual = await vi.importActual("@openstatus/tinybird");
   return {
     // @ts-ignore
     ...actual,
-    publishPingResponse: vi.fn(),
+    publishPingResponse: vi.fn().mockResolvedValue({ successful_rows: 1 }),
   };
 });
 
 it("should call updateMonitorStatus when we can fetch", async () => {
   const spyOn = vi.spyOn(alerts, "updateMonitorStatus").mockReturnThis();
-  await checker({
+  await checkerRetryPolicy({
     workspaceId: "1",
     monitorId: "1",
     url: "https://www.google.com",
···
 it("should call updateMonitorStatus when status error", async () => {
   const spyOn = vi.spyOn(alerts, "updateMonitorStatus").mockReturnThis();
   try {
-    await checker({
+    await checkerRetryPolicy({
       workspaceId: "1",
       monitorId: "1",
       url: "https://xxxxxxx.fake",
···
   } catch (e) {
     expect(e).toBeInstanceOf(Error);
   }
-  expect(spyOn).toHaveBeenCalledTimes(0);
+  expect(spyOn).toHaveBeenCalledTimes(1);
 });
 
 it("What should we do when redirect ", async () => {
   const spyOn = vi.spyOn(alerts, "updateMonitorStatus").mockReturnThis();
+
   try {
-    await checker({
+    await checkerRetryPolicy({
       workspaceId: "1",
       monitorId: "1",
       url: "https://www.openstatus.dev/toto",

apps/server/src/checker/checker.ts

···
-import { nanoid } from "nanoid";
-
-import { publishPingResponse } from "@openstatus/tinybird";
-
-import { env } from "../env";
 import { updateMonitorStatus } from "./alerting";
+import type { PublishPingType } from "./ping";
+import { pingEndpoint, publishPing } from "./ping";
 import type { Payload } from "./schema";
 
-const region = env.FLY_REGION;
-
-export const monitor = async ({
-  monitorInfo,
+// we could have a 'retry' parameter to know how often we should retry
+// we could use a setTimeout to retry after a certain amount of time - can be random between 500ms and 10s
+export const publishPingRetryPolicy = async ({
+  payload,
   latency,
   statusCode,
-}: {
-  monitorInfo: Payload;
-  latency: number;
-  statusCode: number;
-}) => {
-  const { monitorId, cronTimestamp, url, workspaceId } = monitorInfo;
-
+}: PublishPingType) => {
+  try {
+    console.log(
+      `try publish ping to tb - attempt 1  ${JSON.stringify(
+        payload,
+      )}  with latency ${latency} and status code ${statusCode}`,
+    );
+    await publishPing({ payload, statusCode, latency });
+  } catch {
+    try {
+      console.log(
+        "try publish ping to tb - attempt 2 ",
+        JSON.stringify(payload),
+      );
+      await publishPing({ payload, statusCode, latency });
+    } catch (e) {
+      throw e;
+    }
+  }
   console.log(
-    `publishing ping response for ${url} with status ${statusCode} and latency ${latency} and monitorId ${monitorId} `,
+    `Successfully published  ${JSON.stringify(
+      payload,
+    )}  with latency ${latency} and status code ${statusCode}`,
   );
-  await publishPingResponse({
-    id: nanoid(), // TBD: we don't need it
-    timestamp: Date.now(),
-    statusCode,
-    latency,
-    region,
-    url,
-    monitorId,
-    cronTimestamp,
-    workspaceId,
-  });
 };
 
-export const checker = async (data: Payload) => {
-  const startTime = Date.now();
-  const res = await ping(data);
-  const endTime = Date.now();
+const run = async (data: Payload, retry?: number | undefined) => {
+  let startTime = 0;
+  let endTime = 0;
+  let res = null;
+  // We are doing these for wrong urls
+  try {
+    startTime = Date.now();
+    res = await pingEndpoint(data);
+    endTime = Date.now();
+  } catch (e) {
+    console.log("error on pingEndpoint", e);
+    endTime = Date.now();
+  }
+
   const latency = endTime - startTime;
   if (res?.ok) {
-    await monitor({ monitorInfo: data, latency, statusCode: res.status });
+    await publishPingRetryPolicy({
+      payload: data,
+      latency,
+      statusCode: res.status,
+    });
     if (data?.status === "error") {
       await updateMonitorStatus({
         monitorId: data.monitorId,
···
       });
     }
   } else {
-    console.log(`first retry for ${data.url} with status ${res?.status}`);
-    const startTime = Date.now();
-    const retry = await ping(data);
-    const endTime = Date.now();
-    const latency = endTime - startTime;
-    if (retry?.ok) {
-      await monitor({ monitorInfo: data, latency, statusCode: retry.status });
-      if (data?.status === "error") {
+    if (retry === 0) {
+      throw new Error(`error on ping for ${data.monitorId}`);
+    }
+    // Store the error on third task retry
+    if (retry === 1) {
+      await publishPingRetryPolicy({
+        payload: data,
+        latency,
+        statusCode: res?.status || 0,
+      });
+      if (data?.status === "active") {
         await updateMonitorStatus({
           monitorId: data.monitorId,
-          status: "active",
+          status: "error",
         });
       }
-    } else {
-      console.log(
-        `error for ${JSON.stringify(data)} with info ${JSON.stringify(retry)}`,
-      );
     }
   }
+  return { res, latency };
 };
 
-export const ping = async (
-  data: Pick<Payload, "headers" | "body" | "method" | "url">,
-) => {
-  const headers =
-    data?.headers?.reduce((o, v) => {
-      if (v.key.trim() === "") return o; // removes empty keys from the header
-      return { ...o, [v.key]: v.value };
-    }, {}) || {};
-
+export const checkerRetryPolicy = async (data: Payload, retry = 0) => {
   try {
-    const res = await fetch(data?.url, {
-      method: data?.method,
-      cache: "no-store",
-      headers: {
-        "OpenStatus-Ping": "true",
-        ...headers,
-      },
-      // Avoid having "TypeError: Request with a GET or HEAD method cannot have a body." error
-      ...(data.method === "POST" && { body: data?.body }),
-    });
-
-    return res;
-  } catch (e) {
-    console.log(`fetch error for : ${data} with error ${e}`);
-    console.log(e);
+    console.log("try run checker - attempt 1 ", JSON.stringify(data));
+    await run(data, 0);
+  } catch {
+    try {
+      console.log("try run checker - attempt 2 ", JSON.stringify(data));
+      await run(data, 1);
+    } catch (e) {
+      throw e;
+    }
   }
+  console.log("successfully run checker ", JSON.stringify(data));
 };

apps/server/src/checker/index.ts

···
 import { Hono } from "hono";
 
 import { env } from "../env";
-import { catchTooManyRetry } from "./alerting";
-import { checker } from "./checker";
+import { checkerRetryPolicy } from "./checker";
 import { payloadSchema } from "./schema";
 import type { Payload } from "./schema";
 
···
     console.error(result.error);
     return c.text("Unprocessable Entity", 422);
   }
-
-  if (Number(c.req.header("X-CloudTasks-TaskRetryCount") || 0) > 5) {
+  const retry = Number(c.req.header("X-CloudTasks-TaskRetryCount") || 0);
+  if (retry > 3) {
     console.error(
       `catchTooManyRetry for ${JSON.stringify(result.data)}
       )}`,
     );
     // catchTooManyRetry(result.data);
-    return c.text("Ok", 200); // needs to be 200, otherwise qstash will retry
+    return c.text("Ok", 200); // finish the task
   }
 
   console.log(`Google Checker should try this: ${result.data.url}`);
···
     console.log(
       `start checker URL: ${result.data.url} monitorId ${result.data.monitorId}`,
     );
-    checker(result.data);
+    await checkerRetryPolicy(result.data, retry);
     console.log(
       `end checker URL: ${result.data.url} monitorId ${result.data.monitorId}`,
     );
     return c.text("Ok", 200);
   } catch (e) {
-    console.error(e);
+    console.error(
+      `fail checker URL: ${result.data.url} monitorId ${result.data.monitorId}`,
+      JSON.stringify(result.data),
+      e,
+    );
     return c.text("Internal Server Error", 500);
   }
 });
···
     console.error(result.error);
     return c.text("Unprocessable Entity", 422);
   }
-
-  if (Number(c.req.header("X-CloudTasks-TaskRetryCount") || 0) > 5) {
+  const retry = Number(c.req.header("X-CloudTasks-TaskRetryCount") || 0);
+  if (retry > 1) {
     console.error(
       `catchTooManyRetry for ${JSON.stringify(result.data)}
       )}`,
     );
     // catchTooManyRetry(result.data);
-    return c.text("Ok", 200); // needs to be 200, otherwise qstash will retry
+    return c.text("Ok", 200); // finish the task
   }
 
   console.log(`Google Checker should try this: ${result.data.url}`);
···
     console.log(
       `start checker URL: ${result.data.url} monitorId ${result.data.monitorId}`,
     );
-    checker(result.data);
+    await checkerRetryPolicy(result.data, retry);
     console.log(
       `end checker URL: ${result.data.url} monitorId ${result.data.monitorId}`,
     );
     return c.text("Ok", 200);
   } catch (e) {
-    console.error(e);
+    console.error(
+      `fail checker URL: ${result.data.url} monitorId ${result.data.monitorId}`,
+      JSON.stringify(result.data),
+      e,
+    );
     return c.text("Internal Server Error", 500);
   }
 });

apps/server/src/checker/ping.ts

···
+import { nanoid } from "nanoid";
+
+import { publishPingResponse } from "@openstatus/tinybird";
+
+import { env } from "../env";
+import { fakePromiseWithRandomResolve } from "../utils/random-promise";
+import type { Payload } from "./schema";
+
+const region = env.FLY_REGION;
+
+function getHeaders(data?: Payload) {
+  const customHeaders =
+    data?.headers?.reduce((o, v) => {
+      // removes empty keys from the header
+      if (v.key.trim() === "") return o;
+      return { ...o, [v.key]: v.value };
+    }, {}) || {};
+  return {
+    "OpenStatus-Ping": "true",
+    ...customHeaders,
+  };
+}
+
+export async function pingEndpoint(data: Payload) {
+  try {
+    const res = await fetch(data?.url, {
+      method: data?.method,
+      keepalive: false,
+      cache: "no-store",
+      headers: getHeaders(data),
+      // Avoid having "TypeError: Request with a GET or HEAD method cannot have a body." error
+      ...(data.method === "POST" && { body: data?.body }),
+    });
+
+    return res;
+  } catch (e) {
+    throw e;
+  }
+}
+
+export type PublishPingType = {
+  payload: Payload;
+  statusCode: number;
+  latency: number;
+};
+
+export async function publishPing({
+  payload,
+  statusCode,
+  latency,
+}: PublishPingType) {
+  const { monitorId, cronTimestamp, url, workspaceId } = payload;
+
+  if (
+    process.env.NODE_ENV === "production" ||
+    process.env.NODE_ENV === "test"
+  ) {
+    const res = await publishPingResponse({
+      id: nanoid(), // TBD: we don't need it
+      timestamp: Date.now(),
+      statusCode,
+      latency,
+      region,
+      url,
+      monitorId,
+      cronTimestamp,
+      workspaceId,
+    });
+    if (res.successful_rows === 0) {
+      throw new Error(`error 0 rows on publish ping for ${payload.monitorId}`);
+    }
+    return res;
+  }
+
+  const res = await fakePromiseWithRandomResolve();
+  return res;
+}

apps/server/src/utils/random-promise.ts

···
+export function fakePromiseWithRandomResolve() {
+  return new Promise((resolve, reject) => {
+    const randomTime = Math.floor(Math.random() * 1000);
+    setTimeout(() => {
+      const shouldResolve = Math.random() < 0; // 0.5
+      if (shouldResolve) {
+        resolve("Promise resolved successfully.");
+      } else {
+        reject(new Error("Promise rejected."));
+      }
+    }, randomTime);
+  });
+}

apps/web/src/app/api/checker/cron/_cron.ts

···
       const [response] = await client.createTask(request);
 
       allResult.push(response);
-
-      /**
-       * Pushing to our Golang endpoint
-       */
-      const tempTask: google.cloud.tasks.v2beta3.ITask = {
-        httpRequest: {
-          headers: {
-            "Content-Type": "application/json", // Set content type to ensure compatibility your application's request parsing
-            ...(region !== "auto" && { "fly-prefer-region": region }), // Specify the region you want the request to be sent to
-            Authorization: `Basic ${env.CRON_SECRET}`,
-          },
-          httpMethod: "POST",
-          url: "https://checker.openstatus.dev/",
-          body: Buffer.from(JSON.stringify(payload)).toString("base64"),
-        },
-      };
-      const tempRequest = { parent: parent, task: tempTask };
-      const [tempResponse] = await client.createTask(tempRequest);
-
-      allResult.push(tempResponse);
     }
   }
   await Promise.all(allResult);