refactor(api): enhance analytics integration by adding datasets to wrangler configuration and updating node types for compute usage tracking

Author: bchapuis

apps/api/src/context.ts

@@ -1,5 +1,6 @@
 import { Workflow } from "@cloudflare/workers-types";
 import { Fetcher } from "@cloudflare/workers-types";
+import { AnalyticsEngineDataset } from "@cloudflare/workers-types";
 import { JWTTokenPayload } from "@dafthunk/types";
 
 import { RuntimeParams } from "./runtime/runtime";
@@ -11,6 +12,7 @@ export interface Bindings {
   AI: Ai;
   KV: KVNamespace;
   BROWSER: Fetcher;
+  COMPUTE: AnalyticsEngineDataset;
   WEB_HOST: string;
   EMAIL_DOMAIN: string;
   JWT_SECRET: string;

apps/api/src/nodes/node-registry.ts

@@ -326,4 +326,12 @@ export class NodeRegistry {
         !nodeType.compatibility || nodeType.compatibility.includes(workflowType)
     );
   }
+
+  public getNodeType(nodeType: string): NodeType {
+    const Implementation = this.implementations.get(nodeType);
+    if (!Implementation) {
+      throw new Error(`Node type not found: ${nodeType}`);
+    }
+    return Implementation.nodeType;
+  }
 }

apps/api/src/nodes/text/llama-3-1-8b-instruct-fast-node.ts

@@ -13,6 +13,7 @@ export class Llama318BInstructFastNode extends ExecutableNode {
     description: "Generates text",
     category: "Text",
     icon: "ai",
+    usage: 10,
     inputs: [
       {
         name: "prompt",

apps/api/src/nodes/text/llama-3-3-70b-instruct-fp8-fast-node.ts

@@ -20,6 +20,7 @@ export class Llama3370BInstructFastNode extends ExecutableNode {
     description: "Generates text using Llama 3.3 70B model with fp8 precision",
     category: "Text",
     icon: "ai",
+    usage: 100,
     inputs: [
       {
         name: "prompt",

apps/api/src/runtime/runtime.ts

@@ -158,6 +158,7 @@ export class Runtime extends WorkflowEntrypoint<Bindings, RuntimeParams> {
           async () =>
             this.executeNode(
               runtimeState,
+              workflow.id,
               nodeIdentifier,
               organizationId,
               instanceId,
@@ -251,6 +252,7 @@ export class Runtime extends WorkflowEntrypoint<Bindings, RuntimeParams> {
    */
   private async executeNode(
     runtimeState: RuntimeState,
+    workflowId: string,
     nodeIdentifier: string,
     organizationId: string,
     executionId: string,
@@ -268,6 +270,15 @@ export class Runtime extends WorkflowEntrypoint<Bindings, RuntimeParams> {
       return { ...runtimeState, status: "error" };
     }
 
+    const nodeType = NodeRegistry.getInstance().getNodeType(node.type);
+    if (nodeType.usage) {
+      this.env.COMPUTE.writeDataPoint({
+        indexes: [organizationId],
+        blobs: [organizationId, workflowId, node.id],
+        doubles: [nodeType.usage],
+      });
+    }
+
     // Resolve the runnable implementation.
     const executable = NodeRegistry.getInstance().createExecutableNode(node);
     if (!executable) {

apps/api/wrangler.jsonc

@@ -63,6 +63,9 @@
       "class_name": "Runtime",
     },
   ],
+  "analytics_engine_datasets": [
+    { "binding": "COMPUTE", "dataset": "dafthunk-compute" },
+  ],
 
   /**
    * Environment Variables
@@ -114,6 +117,9 @@
           "preview_id": "dd17ff67a75c41768dc30078eec37c46",
         },
       ],
+      "analytics_engine_datasets": [
+        { "binding": "COMPUTE", "dataset": "dafthunk-compute" },
+      ],
       "vars": {
         "WEB_HOST": "https://www.dafthunk.com",
       },

packages/types/src/workflow.ts

@@ -99,6 +99,7 @@ export interface NodeType {
   description?: string;
   category: string;
   icon: string;
+  usage?: number;
   inputs: Parameter[];
   outputs: Parameter[];
   compatibility?: WorkflowType[]; // Optional array of workflow types this node is compatible with