feat: audit logs (#479)

* feat: tb audit log

* chore: add example

* fix: pipe order by

* feat: add generic metadata schema

* chore: audit example in server

* chore: more zod magic

* wip: audit log

* fix: import and descriptions

* chore: improve audit log

* chore: include types

* fix: typo

* chore: small improvements

* fix: index

apps/server/src/checker/alerting.ts

@@ -7,6 +7,7 @@ import {
 import { flyRegionsDict } from "@openstatus/utils";
 
 import { env } from "../env";
+import { checkerAudit } from "../utils/audit-log";
 import { providerToFunction } from "./utils";
 
 export const triggerAlerting = async ({
@@ -43,6 +44,14 @@ export const triggerAlerting = async ({
       statusCode,
       message,
     });
+    // ALPHA
+    await checkerAudit.publishAuditLog({
+      id: `monitor:${monitorId}`,
+      action: "notification.sent",
+      targets: [{ id: monitorId, type: "monitor" }],
+      metadata: { provider: notif.notification.provider },
+    });
+    //
   }
 };
 

apps/server/src/checker/checker.ts

@@ -1,5 +1,4 @@
-import { env } from "../env";
-import { triggerAlerting, upsertMonitorStatus } from "./alerting";
+import { handleMonitorFailed, handleMonitorRecovered } from "./monitor-handler";
 import type { PublishPingType } from "./ping";
 import { pingEndpoint, publishPing } from "./ping";
 import type { Payload } from "./schema";
@@ -76,10 +75,7 @@ const run = async (data: Payload, retry: number) => {
       message: undefined,
     });
     if (data?.status === "error") {
-      await upsertMonitorStatus({
-        monitorId: data.monitorId,
-        status: "active",
-      });
+      handleMonitorRecovered(data, res);
     }
   } else {
     if (retry < 2) {
@@ -96,20 +92,10 @@ const run = async (data: Payload, retry: number) => {
         payload: data,
         latency,
         statusCode: res?.status,
-        message: message,
+        message,
       });
-
       if (data?.status === "active") {
-        await upsertMonitorStatus({
-          monitorId: data.monitorId,
-          status: "error",
-        });
-        await triggerAlerting({
-          monitorId: data.monitorId,
-          region: env.FLY_REGION,
-          statusCode: res?.status,
-          message,
-        });
+        handleMonitorFailed(data, res, message);
       }
     }
   }

apps/server/src/checker/monitor-handler.ts

@@ -0,0 +1,48 @@
+import { env } from "../env";
+import { checkerAudit } from "../utils/audit-log";
+import { triggerAlerting, upsertMonitorStatus } from "./alerting";
+import type { Payload } from "./schema";
+
+export async function handleMonitorRecovered(data: Payload, res: Response) {
+  await upsertMonitorStatus({
+    monitorId: data.monitorId,
+    status: "active",
+  });
+  // ALPHA
+  await checkerAudit.publishAuditLog({
+    id: `monitor:${data.monitorId}`,
+    action: "monitor.recovered",
+    targets: [{ id: data.monitorId, type: "monitor" }],
+    metadata: { region: env.FLY_REGION, statusCode: res.status },
+  });
+  //
+}
+
+export async function handleMonitorFailed(
+  data: Payload,
+  res: Response | null,
+  message?: string,
+) {
+  await upsertMonitorStatus({
+    monitorId: data.monitorId,
+    status: "error",
+  });
+  // ALPHA
+  await checkerAudit.publishAuditLog({
+    id: `monitor:${data.monitorId}`,
+    action: "monitor.failed",
+    targets: [{ id: data.monitorId, type: "monitor" }],
+    metadata: {
+      region: env.FLY_REGION,
+      statusCode: res?.status,
+      message,
+    },
+  });
+  //
+  await triggerAlerting({
+    monitorId: data.monitorId,
+    region: env.FLY_REGION,
+    statusCode: res?.status,
+    message,
+  });
+}

apps/server/src/utils/audit-log.ts

@@ -0,0 +1,7 @@
+import { AuditLog, Tinybird } from "@openstatus/tinybird";
+
+import { env } from "../env";
+
+const tb = new Tinybird({ token: env.TINY_BIRD_API_KEY });
+
+export const checkerAudit = new AuditLog({ tb });

packages/tinybird/datasources/audit_log.datasource

@@ -0,0 +1,13 @@
+VERSION 0
+
+SCHEMA >
+    `action` String `json:$.action`,
+    `actor` String `json:$.actor`,
+    `id` String `json:$.id`,
+    `targets` Nullable(String) `json:$.targets`,
+    `metadata` Nullable(String) `json:$.metadata`,
+    `timestamp` Int64 `json:$.timestamp`,
+    `version` Int16 `json:$.version`
+
+ENGINE "MergeTree"
+ENGINE_SORTING_KEY "id, timestamp, action"

packages/tinybird/package.json

@@ -4,7 +4,7 @@
   "main": "src/index.ts",
   "license": "MIT",
   "dependencies": {
-    "@chronark/zod-bird": "0.2.2",
+    "@chronark/zod-bird": "0.3.1",
     "zod": "3.22.2"
   },
   "devDependencies": {

packages/tinybird/pipes/endpoint_audit_log.pipe

@@ -0,0 +1,8 @@
+VERSION 0
+
+NODE endpoint_audit_pipe_0
+SQL >
+
+    % SELECT * FROM audit_log__v0 WHERE id = {{ String(event_id, 1) }} ORDER BY timestamp DESC
+
+

packages/tinybird/src/audit-log/README.md

@@ -0,0 +1,99 @@
+## Motivation
+
+We want to track every change made for the `incident` and `monitor`. Therefore,
+it requires us to build some audit log / event sourcing foundation.
+
+The `Event` is what the data type stored within [Tinybird](https://tinybird.co).
+It has basic props that every event includes, as well as a `metadata` prop that
+can be used to store additional informations.
+
+```ts
+export type Event = {
+  /**
+   * Unique identifier for the event.
+   */
+  id: string;
+
+  /**
+   * The actor that triggered the event.
+   * @default { id: "", name: "system" }
+   * @example { id: "1", name: "mxkaske" }
+   */
+  actor?: {
+    id: string;
+    name: string;
+  };
+
+  /**
+   * The ressources affected by the action taken.
+   * @example [{ id: "1", name: "monitor" }]
+   */
+  targets?: {
+    id: string;
+    name: string;
+  }[];
+
+  /**
+   * The action that was triggered.
+   * @example monitor.down | incident.create
+   */
+  action: string;
+
+  /**
+   * The timestamp of the event in milliseconds since epoch UTC.
+   * @default Date.now()
+   */
+  timestamp?: number;
+
+  /**
+   * The version of the event. Should be incremented on each update.
+   * @default 1
+   */
+  version?: number;
+
+  /**
+   * Metadata for the event. Defined via zod schema.
+   */
+  metadata?: unknown;
+};
+```
+
+The objects are parsed and stored as string via
+`schema.transform(val => JSON.stringify(val))` and transformed back into an
+object before parsing via `z.preprocess(val => JSON.parse(val), schema)`.
+
+## Example
+
+```ts
+const tb = new Tinybird({ token: process.env.TINY_BIRD_API_KEY || "" });
+
+const auditLog = new AuditLog({ tb });
+
+await auditLog.publishAuditLog({
+  id: "monitor:1",
+  action: "monitor.down",
+  targets: [{ id: "1", type: "monitor" }], // not mandatory, but could be useful later on
+  metadata: { region: "gru", statusCode: 400, message: "timeout" },
+});
+
+await auditLog.getAuditLog({ event_id: "monitor:1" });
+```
+
+## Inspiration
+
+- WorkOS [Audit Logs](https://workos.com/docs/audit-logs)
+
+## Tinybird
+
+Push the pipe and datasource to tinybird:
+
+```
+tb push datasources/audit_log.datasource
+tb push pipes/endpoint_audit_log.pipe
+```
+
+---
+
+### Possible extention
+
+> TODO: Remove `Nullable` from `targets` to better index and query it.

packages/tinybird/src/audit-log/action-schema.ts

@@ -0,0 +1,47 @@
+import { z } from "zod";
+
+/**
+ * The schema for the monitor.recovered action.
+ * It represents the event when a monitor has recovered from a failure.
+ */
+export const monitorRecoveredSchema = z.object({
+  action: z.literal("monitor.recovered"),
+  metadata: z.object({ region: z.string(), statusCode: z.number() }),
+});
+
+/**
+ * The schema for the monitor.failed action.
+ * It represents the event when a monitor has failed.
+ */
+export const monitorFailedSchema = z.object({
+  action: z.literal("monitor.failed"),
+  metadata: z.object({
+    region: z.string(),
+    statusCode: z.number().optional(),
+    message: z.string().optional(),
+  }),
+});
+
+/**
+ * The schema for the notification.send action.
+ *
+ */
+export const notificationSentSchema = z.object({
+  action: z.literal("notification.sent"),
+  // we could use the notificationProviderSchema for more type safety
+  metadata: z.object({ provider: z.string() }),
+});
+
+// TODO: update schemas with correct metadata and description
+
+export const incidentCreatedSchema = z.object({
+  action: z.literal("incident.created"),
+  metadata: z.object({}), // tbd
+});
+
+export const incidentResolvedSchema = z.object({
+  action: z.literal("incident.resolved"),
+  metadata: z.object({}), // tbd
+});
+
+// ...

packages/tinybird/src/audit-log/action-validation.ts

@@ -0,0 +1,67 @@
+import { z } from "zod";
+
+import {
+  monitorFailedSchema,
+  monitorRecoveredSchema,
+  notificationSentSchema,
+} from "./action-schema";
+import { ingestBaseEventSchema, pipeBaseResponseData } from "./base-validation";
+
+/**
+ * The schema for the event object.
+ * It extends the base schema. It uses the `discriminatedUnion` method for faster
+ * evaluation to determine which schema to be used to parse the input.
+ * It also transforms the metadata object into a string.
+ *
+ * @todo: whenever a new action is added, it should be included to the discriminatedUnion
+ */
+export const ingestActionEventSchema = z
+  .intersection(
+    // Unfortunately, the array cannot be dynamic, otherwise could be added to the Client
+    // and made available to devs as library
+    z.discriminatedUnion("action", [
+      monitorRecoveredSchema,
+      monitorFailedSchema,
+      notificationSentSchema,
+    ]),
+    ingestBaseEventSchema,
+  )
+  .transform((val) => ({
+    ...val,
+    metadata: JSON.stringify(val.metadata),
+  }));
+
+/**
+ * The schema for the response object.
+ * It extends the base schema. It uses the `discriminatedUnion` method for faster
+ * evaluation to determine which schema to be used to parse the input.
+ * It also preprocesses the metadata string into the correct schema object.
+ *
+ * @todo: whenever a new action is added, it should be included to the discriminatedUnion
+ */
+export const pipeActionResponseData = z.intersection(
+  z.discriminatedUnion("action", [
+    monitorRecoveredSchema.extend({
+      metadata: z.preprocess(
+        (val) => JSON.parse(String(val)),
+        monitorRecoveredSchema.shape.metadata,
+      ),
+    }),
+    monitorFailedSchema.extend({
+      metadata: z.preprocess(
+        (val) => JSON.parse(String(val)),
+        monitorFailedSchema.shape.metadata,
+      ),
+    }),
+    notificationSentSchema.extend({
+      metadata: z.preprocess(
+        (val) => JSON.parse(String(val)),
+        notificationSentSchema.shape.metadata,
+      ),
+    }),
+  ]),
+  pipeBaseResponseData,
+);
+
+export type IngestActionEvent = z.infer<typeof ingestActionEventSchema>;
+export type PipeActionResponseData = z.infer<typeof pipeActionResponseData>;