construct proper kv step response and send it back from host function (#74)

* construct proper kv step response and send it back from host function

* remove commented code

* add sandbox kv example for reference

* flesh out kv example

---------

Co-authored-by: Jacob Heric <jacob@jacobheric.com:q!>

Author: jacobheric

src/internal/kv.ts

@@ -1,8 +1,14 @@
 import { KVAction } from "@streamdal/protos/protos/shared/sp_shared";
 import { KVCommand } from "@streamdal/protos/protos/sp_command";
 import { KVInstruction } from "@streamdal/protos/protos/sp_kv";
+import {
+  KVStatus,
+  KVStep,
+  KVStepResponse,
+} from "@streamdal/protos/protos/steps/sp_steps_kv";
 
 import { internal } from "./register.js";
+import { writeResponse } from "./wasm.js";
 
 export const kvInstruction = (instruction: KVInstruction) => {
   switch (instruction.action) {
@@ -36,11 +42,29 @@ export const kvCommand = (command: KVCommand) => {
 };
 
 export const kvExists = (
-  memory: WebAssembly.Memory,
-  pointer: number,
-  length: number
-): boolean => {
-  const bytes = new Uint8Array(memory.buffer, pointer, length);
-  const key = new TextDecoder().decode(bytes);
-  return internal.kv.has(key);
+  exports: any,
+  keyPointer: number,
+  keyLength: number
+): bigint => {
+  const kvStep = KVStep.fromBinary(
+    new Uint8Array(exports.memory.buffer, keyPointer, keyLength)
+  );
+  const resultBytes = KVStepResponse.toBinary(
+    KVStepResponse.create({
+      status: internal.kv.has(kvStep.key)
+        ? KVStatus.KV_STATUS_SUCCESS
+        : KVStatus.KV_STATUS_FAILURE,
+    })
+  );
+
+  const resultPointer = exports.alloc(resultBytes.length);
+  const mem = new Uint8Array(
+    exports.memory.buffer,
+    resultPointer,
+    resultBytes.length
+  );
+  mem.set(resultBytes);
+  const response = writeResponse(resultPointer, resultBytes.length);
+
+  return response;
 };

src/internal/wasm.ts

@@ -38,20 +38,8 @@ export const instantiateWasm = async (
   const instantiated = await WebAssembly.instantiate(wasm, {
     wasi_snapshot_preview1: wasi.wasiImport,
     env: {
-      kvExists: (pointer: number, length: number): number => {
-        const result = instantiated.exports.memory
-          ? kvExists(
-              instantiated.exports.memory as WebAssembly.Memory,
-              pointer,
-              length
-            )
-          : false;
-        //
-        // Wasm function expects a bigint but our runtime type expects a number
-        // eslint-disable-next-line @typescript-eslint/ban-ts-comment
-        // @ts-expect-error
-        return BigInt(result ? 1 : 0);
-      },
+      kvExists: (pointer: number, length: number): bigint =>
+        kvExists(instantiated.exports, pointer, length),
     },
   });
   internal.wasmModules.set(wasmId, instantiated);
@@ -68,6 +56,16 @@ export const readResponse = (pointer: bigint, buffer: Uint8Array): any => {
   return buffer.slice(start, start + length);
 };
 
+export const writeResponse = (pointer: number, length: number): bigint => {
+  //
+  // Left shift the pointer value by 32 bits
+  const start = BigInt(pointer) << BigInt(32);
+
+  //
+  // Combine the shifted start and length using bitwise OR
+  return start | BigInt(length);
+};
+
 export const runWasm = ({
   step,
   originalData,

src/sandbox/billing.ts

@@ -83,7 +83,7 @@ const parseJson = (d: string, i: number): any => {
   return null;
 };
 
-const loadData = (path: string): any[] => {
+export const loadData = (path: string): any[] => {
   try {
     const data = readFileSync(path).toString();
 

src/sandbox/index.ts

@@ -112,10 +112,12 @@ const logPipeline = async (streamdal: any, audience: Audience, input: any) => {
       audience.operationType
     ].toLowerCase()}`
   );
-  const { error, errorMessage, data } = await streamdal.process({
-    audience: audience,
-    data: new TextEncoder().encode(JSON.stringify(input)),
-  });
+  const { error, errorMessage, data, pipelineStatus } = await streamdal.process(
+    {
+      audience: audience,
+      data: new TextEncoder().encode(JSON.stringify(input)),
+    }
+  );
   //
   // no active pipeline messages are technically errors
   // but more informational
@@ -129,6 +131,8 @@ const logPipeline = async (streamdal: any, audience: Audience, input: any) => {
   } catch (e) {
     console.error("could not parse data", e);
   }
+  console.debug("pipeline status:");
+  console.dir(pipelineStatus, { depth: 20 });
   console.debug("pipeline request done", new Date());
   console.debug("--------------------------------");
   console.debug("\n");

src/sandbox/kv.ts

@@ -0,0 +1,65 @@
+import { Audience } from "@streamdal/protos/protos/sp_common";
+
+import { OperationType, Streamdal, StreamdalConfigs } from "../streamdal.js";
+import { loadData } from "./billing.js";
+
+const serviceKVConfig: StreamdalConfigs = {
+  streamdalUrl: "localhost:8082",
+  streamdalToken: "1234",
+  serviceName: "kv-service",
+  pipelineTimeout: "100",
+  stepTimeout: "10",
+  dryRun: false,
+};
+
+const KVProducer: Audience = {
+  serviceName: "kv-service",
+  componentName: "postgresql",
+  operationType: OperationType.PRODUCER,
+  operationName: "import",
+};
+
+/**
+ * 1. run this
+ * 2. go to the console, create a pipeline with a "Key/Value" step type,
+ *    choose type "Exists", mode "Dynamic", and use "event_id" for key.
+ * 3. Add the following key via the server rest api:
+ *
+ *    curl --header "Content-Type: application/json" \
+ *      -H "Authorization: Bearer 1234" \
+ *      --request PUT \
+ *      --data '{"kvs": [{"key": "eaab67a7-f8af-48b9-b65f-1f0f15de9956","value": "eaab67a7-f8af-48b9-b65f-1f0f15de9956"}]}' \
+ *      "http://localhost:8081/api/v1/kv"
+ */
+export const kv = () => {
+  const kvService = new Streamdal(serviceKVConfig);
+  const kvData = loadData("./src/sandbox/assets/sample-welcome-producer.json");
+
+  //
+  // eslint-disable-next-line @typescript-eslint/no-misused-promises
+  setInterval(async () => {
+    const result = await kvService.process({
+      audience: KVProducer,
+      data: new TextEncoder().encode(JSON.stringify(kvData[0])),
+    });
+
+    //
+    // Key exists, this will result in a pipeline step running without error
+    // if this is part of multi-step or multi-pipeline you will need to inspect pipelineStatus
+    console.debug(result.error);
+  }, 1000);
+
+  //
+  // eslint-disable-next-line @typescript-eslint/no-misused-promises
+  setInterval(async () => {
+    const result = await kvService.process({
+      audience: KVProducer,
+      data: new TextEncoder().encode(JSON.stringify(kvData[1])),
+    });
+
+    //
+    // Key does not exist, this will result in an error
+    // if this is part of multi-step or multi-pipeline you will need to inspect pipelineStatus
+    console.debug(result.error);
+  }, 1000);
+};