Add Telemetry (#49)

* fix: pass all params to context in failureFunction

* feat: add workflow telemetry

* fix: rm process from hono

* fix: add disable telemetry option and change platform to telemetry

Author: CahidArda

platforms/astro.ts

@@ -1,7 +1,8 @@
 import type { APIContext, APIRoute } from "astro";
 
-import type { PublicServeOptions, WorkflowContext } from "../src";
+import { PublicServeOptions, WorkflowContext } from "../src";
 import { serveBase } from "../src/serve";
+import { SDK_TELEMETRY } from "../src/constants";
 
 export function serve<TInitialPayload = unknown>(
   routeFunction: (
@@ -13,6 +14,13 @@ export function serve<TInitialPayload = unknown>(
   const POST: APIRoute = (apiContext) => {
     const { handler } = serveBase<TInitialPayload>(
       (workflowContext) => routeFunction(workflowContext, apiContext),
+      {
+        sdk: SDK_TELEMETRY,
+        framework: "astro",
+        runtime: process.versions.bun
+          ? `bun@${process.versions.bun}/node@${process.version}`
+          : `node@${process.version}`,
+      },
       options
     );
 

platforms/cloudflare.ts

@@ -1,4 +1,5 @@
 import type { PublicServeOptions, RouteFunction } from "../src";
+import { SDK_TELEMETRY } from "../src/constants";
 import { serveBase } from "../src/serve";
 
 export type WorkflowBindings = {
@@ -62,10 +63,17 @@ export const serve = <TInitialPayload = unknown>(
 ): { fetch: (...args: PagesHandlerArgs | WorkersHandlerArgs) => Promise<Response> } => {
   const fetch = async (...args: PagesHandlerArgs | WorkersHandlerArgs) => {
     const { request, env } = getArgs(args);
-    const { handler: serveHandler } = serveBase(routeFunction, {
-      env,
-      ...options,
-    });
+    const { handler: serveHandler } = serveBase(
+      routeFunction,
+      {
+        sdk: SDK_TELEMETRY,
+        framework: "cloudflare",
+      },
+      {
+        env,
+        ...options,
+      }
+    );
     return await serveHandler(request);
   };
   return { fetch };

platforms/express.ts

@@ -1,4 +1,5 @@
 import type { WorkflowServeOptions, RouteFunction } from "../src";
+import { SDK_TELEMETRY } from "../src/constants";
 import { serveBase } from "../src/serve";
 import {
   Request as ExpressRequest,
@@ -43,7 +44,14 @@ export function serve<TInitialPayload = unknown>(
 
     // create handler
     const { handler: serveHandler } = serveBase<TInitialPayload>(
-      (workflowContext) => routeFunction(workflowContext),
+      routeFunction,
+      {
+        sdk: SDK_TELEMETRY,
+        framework: "express",
+        runtime: process.versions.bun
+          ? `bun@${process.versions.bun}/node@${process.version}`
+          : `node@${process.version}`,
+      },
       {
         ...options,
         useJSONContent: true,

platforms/h3.ts

@@ -3,6 +3,7 @@ import { defineEventHandler, readRawBody } from "h3";
 import type { PublicServeOptions, RouteFunction } from "../src";
 import { serveBase } from "../src/serve";
 import type { IncomingHttpHeaders } from "node:http";
+import { SDK_TELEMETRY } from "../src/constants";
 
 function transformHeaders(headers: IncomingHttpHeaders): [string, string][] {
   const formattedHeaders = Object.entries(headers).map(([key, value]) => [
@@ -37,7 +38,17 @@ export const serve = <TInitialPayload = unknown>(
       method: "POST",
     });
 
-    const { handler: serveHandler } = serveBase<TInitialPayload>(routeFunction, options);
+    const { handler: serveHandler } = serveBase<TInitialPayload>(
+      routeFunction,
+      {
+        sdk: SDK_TELEMETRY,
+        framework: "h3",
+        runtime: process.versions.bun
+          ? `bun@${process.versions.bun}/node@${process.version}`
+          : `node@${process.version}`,
+      },
+      options
+    );
     return await serveHandler(request);
   });
 

platforms/hono.ts

@@ -2,6 +2,7 @@ import type { Context } from "hono";
 import type { PublicServeOptions, RouteFunction } from "../src";
 import { serveBase } from "../src/serve";
 import { Variables } from "hono/types";
+import { SDK_TELEMETRY } from "../src/constants";
 
 export type WorkflowBindings = {
   QSTASH_TOKEN: string;
@@ -32,12 +33,19 @@ export const serve = <
     const environment = context.env;
     const request = context.req.raw;
 
-    const { handler: serveHandler } = serveBase(routeFunction, {
-      // when hono is used without cf workers, it sends a DebugHTTPServer
-      // object in `context.env`. don't pass env if this is the case:
-      env: "QSTASH_TOKEN" in environment ? environment : undefined,
-      ...options,
-    });
+    const { handler: serveHandler } = serveBase(
+      routeFunction,
+      {
+        sdk: SDK_TELEMETRY,
+        framework: "hono",
+      },
+      {
+        // when hono is used without cf workers, it sends a DebugHTTPServer
+        // object in `context.env`. don't pass env if this is the case:
+        env: "QSTASH_TOKEN" in environment ? environment : undefined,
+        ...options,
+      }
+    );
     return await serveHandler(request);
   };
   return handler;

platforms/nextjs.ts

@@ -4,6 +4,7 @@ import type { NextApiHandler, NextApiRequest, NextApiResponse } from "next";
 
 import type { RouteFunction, PublicServeOptions } from "../src";
 import { serveBase } from "../src/serve";
+import { SDK_TELEMETRY } from "../src/constants";
 
 /**
  * Serve method to serve a Upstash Workflow in a Nextjs project
@@ -20,6 +21,11 @@ export const serve = <TInitialPayload = unknown>(
 ): { POST: (request: Request) => Promise<Response> } => {
   const { handler: serveHandler } = serveBase<TInitialPayload, Request, Response>(
     routeFunction,
+    {
+      sdk: SDK_TELEMETRY,
+      framework: "nextjs",
+      runtime: `node@${process.version}`,
+    },
     options
   );
 
@@ -34,7 +40,17 @@ export const servePagesRouter = <TInitialPayload = unknown>(
   routeFunction: RouteFunction<TInitialPayload>,
   options?: PublicServeOptions<TInitialPayload>
 ): { handler: NextApiHandler } => {
-  const { handler: serveHandler } = serveBase(routeFunction, options);
+  const { handler: serveHandler } = serveBase(
+    routeFunction,
+    {
+      sdk: SDK_TELEMETRY,
+      framework: "nextjs-pages",
+      runtime: process.versions.bun
+        ? `bun@${process.versions.bun}/node@${process.version}`
+        : `node@${process.version}`,
+    },
+    options
+  );
 
   const handler = async (request_: NextApiRequest, res: NextApiResponse) => {
     if (request_.method?.toUpperCase() !== "POST") {

platforms/solidjs.ts

@@ -2,6 +2,7 @@ import type { APIEvent } from "@solidjs/start/server";
 
 import type { PublicServeOptions, RouteFunction } from "../src";
 import { serveBase } from "../src/serve";
+import { SDK_TELEMETRY } from "../src/constants";
 
 /**
  * Serve method to serve a Upstash Workflow in a Nextjs project
@@ -28,7 +29,17 @@ export const serve = <TInitialPayload = unknown>(
     }
 
     // create serve handler
-    const { handler: serveHandler } = serveBase<TInitialPayload>(routeFunction, options);
+    const { handler: serveHandler } = serveBase<TInitialPayload>(
+      routeFunction,
+      {
+        sdk: SDK_TELEMETRY,
+        framework: "solidjs",
+        runtime: process.versions.bun
+          ? `bun@${process.versions.bun}/node@${process.version}`
+          : `node@${process.version}`,
+      },
+      options
+    );
 
     return await serveHandler(event.request);
   };

platforms/svelte.ts

@@ -2,6 +2,7 @@ import type { RequestHandler } from "@sveltejs/kit";
 
 import type { PublicServeOptions, RouteFunction } from "../src";
 import { serveBase } from "../src/serve";
+import { SDK_TELEMETRY } from "../src/constants";
 
 /**
  * Serve method to serve a Upstash Workflow in a Nextjs project
@@ -19,10 +20,17 @@ export const serve = <TInitialPayload = unknown>(
   }
 ): { POST: RequestHandler } => {
   const handler: RequestHandler = async ({ request }) => {
-    const { handler: serveHandler } = serveBase<TInitialPayload>(routeFunction, {
-      ...options,
-      useJSONContent: true,
-    });
+    const { handler: serveHandler } = serveBase<TInitialPayload>(
+      routeFunction,
+      {
+        sdk: SDK_TELEMETRY,
+        framework: "svelte",
+      },
+      {
+        ...options,
+        useJSONContent: true,
+      }
+    );
     return await serveHandler(request);
   };
 

src/client/index.ts

@@ -4,7 +4,6 @@ import { makeGetWaitersRequest, makeNotifyRequest } from "./utils";
 import { getWorkflowRunId } from "../utils";
 import { triggerFirstInvocation } from "../workflow-requests";
 import { WorkflowContext } from "../context";
-import { DEFAULT_RETRIES } from "../constants";
 
 type ClientConfig = ConstructorParameters<typeof QStashClient>[0];
 
@@ -214,8 +213,13 @@ export class Client {
       steps: [],
       url,
       workflowRunId: finalWorkflowRunId,
+      retries,
+      telemetry: undefined, // can't know workflow telemetry here
+    });
+    const result = await triggerFirstInvocation({
+      workflowContext: context,
+      telemetry: undefined, // can't know workflow telemetry here
     });
-    const result = await triggerFirstInvocation(context, retries ?? DEFAULT_RETRIES);
     if (result.isOk()) {
       return { workflowRunId: finalWorkflowRunId };
     } else {

src/constants.ts

@@ -1,3 +1,5 @@
+import { Telemetry } from "./types";
+
 export const WORKFLOW_ID_HEADER = "Upstash-Workflow-RunId";
 export const WORKFLOW_INIT_HEADER = "Upstash-Workflow-Init";
 export const WORKFLOW_URL_HEADER = "Upstash-Workflow-Url";
@@ -12,3 +14,15 @@ export const DEFAULT_CONTENT_TYPE = "application/json";
 export const NO_CONCURRENCY = 1;
 export const NOT_SET = "not-set";
 export const DEFAULT_RETRIES = 3;
+
+export const VERSION = "v0.2.3";
+export const SDK_TELEMETRY = `@upstash/workflow@${VERSION}`;
+
+export const TELEMETRY_HEADER_SDK = "Upstash-Telemetry-Sdk" as const;
+export const TELEMETRY_HEADER_FRAMEWORK = "Upstash-Telemetry-Framework" as const;
+export const TELEMETRY_HEADER_RUNTIME = "Upstash-Telemetry-Runtime" as const;
+
+export const MOCK_TELEMETRY: Telemetry = {
+  framework: "mock",
+  sdk: "mock",
+};