feat: Add key-value storage support for plugins (#463)

* feat: Add key-value storage support for plugins

Signed-off-by: Dylan Kilkenny <[email protected]>

* chore: Add comments

Signed-off-by: Dylan Kilkenny <[email protected]>

* refactor: Apply review changes

Signed-off-by: Dylan Kilkenny <[email protected]>

* test: Fix

Signed-off-by: Dylan Kilkenny <[email protected]>

* fix: Rename `scan` method to `listKeys`

---------

Signed-off-by: Dylan Kilkenny <[email protected]>

Author: dylankilkenny

docs/modules/ROOT/pages/plugins.adoc

@@ -1,4 +1,3 @@
-
 ## Relayer Plugins
 
 Relayer plugins are TypeScript functions that can be invoked through the relayer HTTP API.
@@ -10,45 +9,63 @@ Under the hood, the relayer will execute the plugin code in a separate process u
 ### 1. Writing your plugin
 
 ```typescript
-import { Speed } from "@openzeppelin/relayer-sdk";
-import { PluginAPI, runPlugin } from "../lib/plugin";
+import { Speed, PluginContext } from '@openzeppelin/relayer-sdk';
 
 type Params = {
-    destinationAddress: string;
+  destinationAddress: string;
+};
+
+type Result = {
+  success: boolean;
+  transactionId: string;
 };
 
-async function example(api: PluginAPI, params: Params): Promise<string> {
-    console.info("Plugin started...");
-    /**
-     * Instances the relayer with the given id.
-     */
-    const relayer = api.useRelayer("sepolia-example");
-
-    /**
-     * Sends an arbitrary transaction through the relayer.
-     */
-    const result = await relayer.sendTransaction({
-        to: params.destinationAddress,
-        value: 1,
-        data: "0x",
-        gas_limit: 21000,
-        speed: Speed.FAST,
-    });
-
-    /*
-    * Waits for the transaction to be mined on chain.
-    */
-    await result.wait();
-
-    return "done!";
+export async function handler(context: PluginContext): Promise<Result> {
+  const { api, params, kv } = context;
+  console.info('Plugin started...');
+
+  const relayer = api.useRelayer('sepolia-example');
+  const result = await relayer.sendTransaction({
+    to: params.destinationAddress,
+    value: 1,
+    data: '0x',
+    gas_limit: 21000,
+    speed: Speed.FAST,
+  });
+
+  // Optional: persist last transaction id
+  await kv.set('last_tx_id', result.id);
+
+  await result.wait();
+  return { success: true, transactionId: result.id };
 }
+```
+
+#### Legacy patterns (deprecated)
 
-/**
- * This is the entry point for the plugin
- */
-runPlugin(example);
+The following patterns are supported for backward compatibility but will be removed in a future version. They do not provide access to the KV store.
+
+```typescript
+// Legacy: runPlugin pattern (deprecated)
+import { PluginAPI, runPlugin } from '../lib/plugin';
+
+async function legacyMain(api: PluginAPI, params: any) {
+  // logic here (no KV access)
+  return 'done!';
+}
+
+runPlugin(legacyMain);
 ```
 
+```typescript
+// Legacy: two-parameter handler (deprecated, no KV)
+import { PluginAPI } from '@openzeppelin/relayer-sdk';
+
+export async function handler(api: PluginAPI, params: any): Promise<any> {
+  // logic here (no KV access)
+  return 'done!';
+}
+```
 
 ### 2. Adding extra dependencies
 
@@ -61,25 +78,17 @@ pnpm add ethers
 And then just import them in your plugin.
 
 ```typescript
-import { ethers } from "ethers";
+import { ethers } from 'ethers';
 ```
 
 ### 3. Adding to config file
 
 - id: The id of the plugin. This is used to call a specific plugin through the HTTP API.
 - path: The path to the plugin file - relative to the `/plugins` folder.
-- timeout (optional): The timeout for the script execution *in seconds*. If not provided, the default timeout of 300 seconds (5 minutes) will be used.
+- timeout (optional): The timeout for the script execution _in seconds_. If not provided, the default timeout of 300 seconds (5 minutes) will be used.
 
 ```yaml
-{
-  "plugins": [
-    {
-      "id": "example",
-      "path": "examples/example.ts",
-      "timeout": 30
-    }
-  ]
-}
+{ 'plugins': [{ 'id': 'example', 'path': 'examples/example.ts', 'timeout': 30 }] }
 ```
 
 ## Usage
@@ -138,3 +147,49 @@ Example response:
   "error": null
 }
 ```
+
+## Key-Value Store (KV)
+
+Plugins have access to a built-in KV store via the `PluginContext.kv` property for persistent state and safe concurrency.
+
+- Uses the same Redis URL as the Relayer (`REDIS_URL`)
+- Keys are namespaced per plugin ID
+- JSON values are supported
+
+```typescript
+import { PluginContext } from '@openzeppelin/relayer-sdk';
+
+export async function handler(context: PluginContext) {
+  const { kv } = context;
+
+  // Set with optional TTL
+  await kv.set('greeting', { text: 'hello' }, { ttlSec: 3600 });
+
+  // Get
+  const v = await kv.get<{ text: string }>('greeting');
+
+  // Atomic update with lock
+  const count = await kv.withLock(
+    'counter',
+    async () => {
+      const cur = (await kv.get<number>('counter')) ?? 0;
+      const next = cur + 1;
+      await kv.set('counter', next);
+      return next;
+    },
+    { ttlSec: 10 }
+  );
+
+  return { v, count };
+}
+```
+
+Available methods:
+
+- `get<T>(key: string): Promise<T | null>`
+- `set(key: string, value: unknown, opts?: { ttlSec?: number }): Promise<boolean>`
+- `del(key: string): Promise<boolean>`
+- `exists(key: string): Promise<boolean>`
+- `listKeys(pattern?: string, batch?: number): Promise<string[]>`
+- `clear(): Promise<number>`
+- `withLock<T>(key: string, fn: () => Promise<T>, opts?: { ttlSec?: number; onBusy?: 'throw' | 'skip' }): Promise<T | null>`

plugins/README.md

@@ -0,0 +1,132 @@
+import { PluginContext } from '../lib/plugin';
+
+/**
+ * Simple KV storage example
+ *
+ * Demonstrates:
+ * - JSON set/get (with optional TTL)
+ * - exists/del
+ * - scan pattern listing
+ * - clear namespace
+ * - withLock for atomic sections
+ *
+ * Usage (params.action):
+ * - 'demo' (default): run a small end-to-end flow
+ * - 'set': { key: string, value: any, ttlSec?: number }
+ * - 'get': { key: string }
+ * - 'exists': { key: string }
+ * - 'del': { key: string }
+ * - 'scan': { pattern?: string, batch?: number }
+ * - 'clear': {}
+ * - 'withLock': { key: string, ttlSec?: number, onBusy?: 'throw' | 'skip' }
+ */
+export async function handler({ kv, params }: PluginContext) {
+  const action = params?.action ?? 'demo';
+
+  switch (action) {
+    case 'set': {
+      const { key, value, ttlSec } = params ?? {};
+      assertString(key, 'key');
+      const ok = await kv.set(key, value, { ttlSec: toInt(ttlSec) });
+      return { ok };
+    }
+
+    case 'get': {
+      const { key } = params ?? {};
+      assertString(key, 'key');
+      const value = await kv.get(key);
+      return { value };
+    }
+
+    case 'exists': {
+      const { key } = params ?? {};
+      assertString(key, 'key');
+      const exists = await kv.exists(key);
+      return { exists };
+    }
+
+    case 'del': {
+      const { key } = params ?? {};
+      assertString(key, 'key');
+      const deleted = await kv.del(key);
+      return { deleted };
+    }
+
+    case 'scan': {
+      const { pattern, batch } = params ?? {};
+      const keys = await kv.listKeys(pattern ?? '*', toInt(batch, 500));
+      return { keys };
+    }
+
+    case 'clear': {
+      const deleted = await kv.clear();
+      return { deleted };
+    }
+
+    case 'withLock': {
+      const { key, ttlSec, onBusy } = params ?? {};
+      assertString(key, 'key');
+      const result = await kv.withLock(
+        key,
+        async () => {
+          // Simulate a small critical section
+          const stamp = Date.now();
+          await kv.set(`example:last-lock:${key}`, { stamp });
+          return { ok: true, stamp };
+        },
+        { ttlSec: toInt(ttlSec, 30), onBusy: onBusy === 'skip' ? 'skip' : 'throw' }
+      );
+      return { result };
+    }
+
+    case 'demo':
+    default: {
+      // 1) Write JSON and read it back
+      await kv.set('example:greeting', { text: 'hello' });
+      const greeting = await kv.get<{ text: string }>('example:greeting');
+
+      // 2) Write a TTL value (won't await expiry here)
+      await kv.set('example:temp', { expires: true }, { ttlSec: 5 });
+
+      // 3) Check existence and delete
+      const existedBefore = await kv.exists('example:to-delete');
+      await kv.set('example:to-delete', { remove: true });
+      const existedAfterSet = await kv.exists('example:to-delete');
+      const deleted = await kv.del('example:to-delete');
+
+      // 4) Scan keys under example:*
+      const list = await kv.listKeys('example:*');
+
+      // 5) Use a lock to protect an update
+      const lockResult = await kv.withLock(
+        'example:lock',
+        async () => {
+          const count = (await kv.get<number>('example:counter')) ?? 0;
+          const next = count + 1;
+          await kv.set('example:counter', next);
+          return next;
+        },
+        { ttlSec: 10, onBusy: 'throw' }
+      );
+
+      return {
+        greeting,
+        ttlKeyWritten: true,
+        existedBefore,
+        existedAfterSet,
+        deleted,
+        scanned: list,
+        lockResult,
+      };
+    }
+  }
+}
+
+function toInt(v: unknown, def = 0): number {
+  const n = typeof v === 'string' ? parseInt(v, 10) : typeof v === 'number' ? Math.floor(v) : def;
+  return Number.isFinite(n) && n > 0 ? n : def;
+}
+
+function assertString(v: any, name: string): asserts v is string {
+  if (typeof v !== 'string' || v.length === 0) throw new Error(`${name} must be a non-empty string`);
+}

plugins/examples/kv-storage.ts

@@ -15,10 +15,11 @@
  * 3. Calls the plugin's exported 'handler' function
  * 4. Returns results back to the Rust environment
  *
- * Usage: ts-node executor.ts <socket_path> <params_json> <user_script_path>
+ * Usage: ts-node executor.ts <socket_path> <plugin_id> <params_json> <user_script_path>
  *
  * Arguments:
  * - socket_path: Unix socket path for communication with relayer
+ * - plugin_id: Plugin ID for namespacing KV storage
  * - params_json: JSON string containing plugin parameters
  * - user_script_path: Path to the user's plugin file to execute
  */
@@ -29,27 +30,33 @@ import { LogInterceptor } from './logger';
 
 /**
  * Extract and validate CLI arguments passed from Rust script_executor.rs
+ * Now includes pluginId as a separate argument
  */
 function extractCliArguments() {
-  // Get arguments: [node, executor.ts, socketPath, paramsJson, userScriptPath]
+  // Get arguments: [node, executor.ts, socketPath, pluginId, paramsJson, userScriptPath]
   const socketPath = process.argv[2];
-  const paramsJson = process.argv[3];
-  const userScriptPath = process.argv[4];
+  const pluginId = process.argv[3]; // NEW: Plugin ID as separate arg
+  const paramsJson = process.argv[4]; // Shifted from argv[3]
+  const userScriptPath = process.argv[5]; // Shifted from argv[4]
 
   // Validate required arguments
   if (!socketPath) {
     throw new Error("Socket path is required (argument 1)");
   }
 
+  if (!pluginId) {
+    throw new Error("Plugin ID is required (argument 2)");
+  }
+
   if (!paramsJson) {
-    throw new Error("Plugin parameters JSON is required (argument 2)");
+    throw new Error("Plugin parameters JSON is required (argument 3)");
   }
 
   if (!userScriptPath) {
-    throw new Error("User script path is required (argument 3)");
+    throw new Error("User script path is required (argument 4)");
   }
 
-  return { socketPath, paramsJson, userScriptPath };
+  return { socketPath, pluginId, paramsJson, userScriptPath };
 }
 
 /**
@@ -74,14 +81,14 @@ async function main(): Promise<void> {
     // This provides better backward compatibility with existing scripts
     logInterceptor.start();
 
-    // Extract and validate CLI arguments
-    const { socketPath, paramsJson, userScriptPath } = extractCliArguments();
+    // Extract and validate CLI arguments including plugin ID
+    const { socketPath, pluginId, paramsJson, userScriptPath } = extractCliArguments();
 
     // Parse plugin parameters
     const pluginParams = parsePluginParameters(paramsJson);
 
-    // Execute plugin with validated parameters
-    const result = await runUserPlugin(socketPath, pluginParams, userScriptPath);
+    // Pass plugin ID as separate argument
+    const result = await runUserPlugin(socketPath, pluginId, pluginParams, userScriptPath);
 
     // Add the result to LogInterceptor output
     logInterceptor.addResult(serializeResult(result));