fix(FN-3794): pass unload context and isolate WhatsApp sessions per project
- Extend PluginOnUnload to receive runtime context and wire ctx through plugin-loader unload hooks - Scope WhatsApp chat plugin connections by project root to avoid cross-project session leakage - Update plugin authoring docs and add a patch changeset for @runfusion/fusion - Align plugin test suites across WhatsApp and example/runtime plugins with the new onUnload context contract Fusion-Task-Id: FN-3794
This commit is contained in:
7
.changeset/fn-3794-whatsapp-multi-project.md
Normal file
7
.changeset/fn-3794-whatsapp-multi-project.md
Normal file
@@ -0,0 +1,7 @@
|
||||
---
|
||||
"@runfusion/fusion": patch
|
||||
---
|
||||
|
||||
Fix a multi-project collision in the bundled WhatsApp plugin by keying connections with `getRootDir() + "::" + pluginId`, so concurrent projects no longer share a single connection state.
|
||||
|
||||
Update the plugin SDK hook type so `onUnload` now receives `PluginContext` (matching `onLoad`). This is backward-compatible at runtime, but plugin authors may need to update TypeScript signatures.
|
||||
@@ -306,7 +306,7 @@ const plugin: FusionPlugin = {
|
||||
| Hook | Signature | When It Fires |
|
||||
|------|-----------|---------------|
|
||||
| `onLoad` | `(ctx: PluginContext) => Promise<void> \| void` | Plugin first loaded and started |
|
||||
| `onUnload` | `() => Promise<void> \| void` | Plugin stopped/shutdown |
|
||||
| `onUnload` | `(ctx: PluginContext) => Promise<void> \| void` | Plugin stopped/shutdown |
|
||||
| `onTaskCreated` | `(task: Task, ctx: PluginContext) => Promise<void> \| void` | New task created |
|
||||
| `onTaskMoved` | `(task: Task, fromColumn: string, toColumn: string, ctx: PluginContext) => Promise<void> \| void` | Task moved between columns |
|
||||
| `onTaskCompleted` | `(task: Task, ctx: PluginContext) => Promise<void> \| void` | Task reached "done" |
|
||||
@@ -315,6 +315,7 @@ const plugin: FusionPlugin = {
|
||||
|
||||
### Hook Behavior
|
||||
|
||||
- **Context parity**: `onUnload` receives the same `PluginContext` shape as `onLoad`.
|
||||
- **Timeout**: 5 seconds per invocation (logged and skipped if exceeded)
|
||||
- **Error Isolation**: Hook failures never block other hooks or abort startup
|
||||
- **Optional**: Only define the hooks you need
|
||||
@@ -1266,8 +1267,9 @@ export default definePlugin({
|
||||
onTaskCreated: (task, ctx) => {
|
||||
ctx.logger.info(`Task created: ${task.id}`);
|
||||
},
|
||||
onUnload: () => {
|
||||
// Cleanup
|
||||
onUnload: (ctx) => {
|
||||
// Cleanup with the same context shape passed to onLoad
|
||||
ctx.logger.info("Shutting down plugin");
|
||||
},
|
||||
},
|
||||
} satisfies FusionPlugin);
|
||||
|
||||
@@ -129,6 +129,8 @@ function droidPluginModulePath(): string {
|
||||
// Mock TaskStore for testing
|
||||
const mockTaskStore = {
|
||||
logActivity: vi.fn(),
|
||||
getRootDir: () => "/tmp/plugin-loader-test-root",
|
||||
getPluginStore: vi.fn(),
|
||||
} as any;
|
||||
|
||||
type MockStructuredLogger = {
|
||||
@@ -785,6 +787,42 @@ export default plugin;
|
||||
expect(loader.isPluginLoaded("remove-test")).toBe(false);
|
||||
});
|
||||
|
||||
it("passes plugin context to onUnload", async () => {
|
||||
await pluginStore.init();
|
||||
|
||||
const plugin = makePlugin(makeManifest({ id: "stop-context-test" }));
|
||||
const pluginDir = join(rootDir, "plugins");
|
||||
const pluginPath = await writePluginWithHooks(
|
||||
pluginDir,
|
||||
"stop-context.js",
|
||||
{
|
||||
onUnload:
|
||||
"(ctx => { globalThis.__pluginUnloadCtx = { pluginId: ctx.pluginId, taskStore: ctx.taskStore }; })",
|
||||
},
|
||||
plugin.manifest,
|
||||
);
|
||||
|
||||
await pluginStore.registerPlugin({
|
||||
manifest: plugin.manifest,
|
||||
path: pluginPath,
|
||||
});
|
||||
|
||||
const loader = new PluginLoader({
|
||||
pluginStore,
|
||||
taskStore: mockTaskStore,
|
||||
});
|
||||
|
||||
await loader.loadPlugin("stop-context-test");
|
||||
await loader.stopPlugin("stop-context-test");
|
||||
|
||||
const unloadCtx = (globalThis as { __pluginUnloadCtx?: { pluginId: string; taskStore: unknown } })
|
||||
.__pluginUnloadCtx;
|
||||
expect(unloadCtx).toBeDefined();
|
||||
expect(unloadCtx?.pluginId).toBe("stop-context-test");
|
||||
expect(unloadCtx?.taskStore).toBe(mockTaskStore);
|
||||
delete (globalThis as { __pluginUnloadCtx?: unknown }).__pluginUnloadCtx;
|
||||
});
|
||||
|
||||
it("no-ops for non-loaded plugin", async () => {
|
||||
await pluginStore.init();
|
||||
|
||||
|
||||
@@ -374,8 +374,9 @@ export class PluginLoader extends EventEmitter<{
|
||||
|
||||
// Call onUnload with timeout
|
||||
try {
|
||||
const ctx = await this.createContext(oldPlugin);
|
||||
await this.withTimeout(
|
||||
this.safeCallHook(oldPlugin, "onUnload", []),
|
||||
this.safeCallHook(oldPlugin, "onUnload", [ctx]),
|
||||
timeoutMs,
|
||||
`onUnload timeout for ${pluginId}`,
|
||||
);
|
||||
@@ -645,8 +646,9 @@ export class PluginLoader extends EventEmitter<{
|
||||
|
||||
try {
|
||||
// Call onUnload hook
|
||||
const ctx = await this.createContext(plugin);
|
||||
await this.withTimeout(
|
||||
this.safeCallHook(plugin, "onUnload", []),
|
||||
this.safeCallHook(plugin, "onUnload", [ctx]),
|
||||
5000,
|
||||
`onUnload timeout for ${pluginId}`,
|
||||
);
|
||||
|
||||
@@ -154,7 +154,7 @@ export interface PluginLogger {
|
||||
/** Lifecycle hook: called when plugin is loaded */
|
||||
export type PluginOnLoad = (ctx: PluginContext) => Promise<void> | void;
|
||||
/** Lifecycle hook: called when plugin is unloaded */
|
||||
export type PluginOnUnload = () => Promise<void> | void;
|
||||
export type PluginOnUnload = (ctx: PluginContext) => Promise<void> | void;
|
||||
/** Lifecycle hook: called during database schema initialization */
|
||||
export type PluginOnSchemaInit = (db: Database) => Promise<void> | void;
|
||||
/** Lifecycle hook: called when a task is created */
|
||||
|
||||
@@ -185,7 +185,7 @@ describe("ci-status plugin", () => {
|
||||
await plugin.hooks.onLoad?.(ctx as any);
|
||||
vi.clearAllMocks();
|
||||
|
||||
await plugin.hooks.onUnload?.();
|
||||
await plugin.hooks.onUnload?.(ctx as any);
|
||||
|
||||
expect(ctx.logger.info).toHaveBeenCalledWith("CI Status plugin unloaded");
|
||||
});
|
||||
|
||||
@@ -51,6 +51,6 @@ describe("even realities plugin", () => {
|
||||
const res = await statusRoute?.handler({}, ctx);
|
||||
|
||||
expect(res).toMatchObject({ status: 200, body: { connected: true } });
|
||||
await plugin.hooks?.onUnload?.();
|
||||
await plugin.hooks?.onUnload?.(ctx);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -144,6 +144,7 @@ describe("openclaw-runtime plugin", () => {
|
||||
});
|
||||
|
||||
it("onUnload does not throw", () => {
|
||||
expect(() => plugin.hooks!.onUnload?.()).not.toThrow();
|
||||
const ctx = createMockContext() as any;
|
||||
expect(() => plugin.hooks!.onUnload?.(ctx)).not.toThrow();
|
||||
});
|
||||
});
|
||||
|
||||
@@ -12,7 +12,7 @@
|
||||
},
|
||||
"scripts": {
|
||||
"build": "tsc",
|
||||
"test": "vitest run --silent=passed-only --reporter=dot"
|
||||
"test": "pnpm dlx vitest run --silent=passed-only --reporter=dot"
|
||||
},
|
||||
"dependencies": {
|
||||
"@fusion/core": "workspace:*",
|
||||
|
||||
@@ -1,5 +1,34 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { describe, expect, it, vi, beforeEach } from "vitest";
|
||||
import type { PluginContext } from "@fusion/plugin-sdk";
|
||||
|
||||
const connectionInstances: Array<{
|
||||
start: ReturnType<typeof vi.fn>;
|
||||
stop: ReturnType<typeof vi.fn>;
|
||||
getStatus: ReturnType<typeof vi.fn>;
|
||||
requestPairingCode: ReturnType<typeof vi.fn>;
|
||||
logout: ReturnType<typeof vi.fn>;
|
||||
}> = [];
|
||||
|
||||
vi.mock("../connection.js", () => {
|
||||
const ctor = vi.fn((ctx: PluginContext) => {
|
||||
const root = ctx.taskStore.getRootDir();
|
||||
const instance = {
|
||||
start: vi.fn(async () => {}),
|
||||
stop: vi.fn(async () => {}),
|
||||
getStatus: vi.fn(() => ({ state: "open", jid: root })),
|
||||
requestPairingCode: vi.fn(async () => "123-456"),
|
||||
logout: vi.fn(async () => {}),
|
||||
};
|
||||
connectionInstances.push(instance);
|
||||
return instance;
|
||||
});
|
||||
(ctor as unknown as { splitMessageForWhatsapp: (text: string) => string[] }).splitMessageForWhatsapp =
|
||||
(text: string) => (text.length > 4096 ? [text.slice(0, 4096), text.slice(4096, 8192), text.slice(8192)] : [text]);
|
||||
return { WhatsAppConnection: ctor };
|
||||
});
|
||||
|
||||
import plugin, { ensureSchema, getDedupeRetentionDays, markProcessed, splitMessageForWhatsapp, wasProcessed } from "../index.js";
|
||||
import { WhatsAppConnection } from "../connection.js";
|
||||
|
||||
function createInMemoryDb() {
|
||||
const dedupe = new Map<string, { sender: string; receivedAt: string }>();
|
||||
@@ -36,6 +65,10 @@ function createInMemoryDb() {
|
||||
}
|
||||
|
||||
describe("whatsapp plugin", () => {
|
||||
beforeEach(() => {
|
||||
connectionInstances.length = 0;
|
||||
vi.clearAllMocks();
|
||||
});
|
||||
it("registers schema init hook", () => {
|
||||
expect(plugin.hooks?.onSchemaInit).toBeDefined();
|
||||
});
|
||||
@@ -67,6 +100,66 @@ describe("whatsapp plugin", () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe("multi-project isolation", () => {
|
||||
it("keeps project contexts isolated with shared plugin id", async () => {
|
||||
const db = createInMemoryDb();
|
||||
const makeCtx = (rootDir: string): PluginContext => ({
|
||||
pluginId: "fusion-plugin-whatsapp-chat",
|
||||
settings: {},
|
||||
logger: {
|
||||
info: vi.fn(),
|
||||
warn: vi.fn(),
|
||||
error: vi.fn(),
|
||||
debug: vi.fn(),
|
||||
},
|
||||
emitEvent: vi.fn(),
|
||||
taskStore: {
|
||||
getRootDir: () => rootDir,
|
||||
getPluginStore: () => ({
|
||||
db,
|
||||
}),
|
||||
} as unknown as PluginContext["taskStore"],
|
||||
});
|
||||
|
||||
const ctxA = makeCtx("/repo-a");
|
||||
const ctxB = makeCtx("/repo-b");
|
||||
|
||||
await plugin.hooks!.onLoad!(ctxA);
|
||||
await plugin.hooks!.onLoad!(ctxB);
|
||||
|
||||
expect(WhatsAppConnection).toHaveBeenCalledTimes(2);
|
||||
expect(connectionInstances[0]?.start).toHaveBeenCalledTimes(1);
|
||||
expect(connectionInstances[1]?.start).toHaveBeenCalledTimes(1);
|
||||
|
||||
const statusRoute = plugin.routes!.find((route) => route.method === "GET" && route.path === "/status")!;
|
||||
|
||||
const statusA = await statusRoute.handler({} as never, ctxA) as { status: number; body: unknown };
|
||||
const statusB = await statusRoute.handler({} as never, ctxB) as { status: number; body: unknown };
|
||||
expect(statusA.status).toBe(200);
|
||||
expect((statusA.body as { jid: string }).jid).toBe("/repo-a");
|
||||
expect(statusB.status).toBe(200);
|
||||
expect((statusB.body as { jid: string }).jid).toBe("/repo-b");
|
||||
|
||||
await plugin.hooks!.onUnload!(ctxA);
|
||||
expect(connectionInstances[0]?.stop).toHaveBeenCalledTimes(1);
|
||||
expect(connectionInstances[1]?.stop).not.toHaveBeenCalled();
|
||||
|
||||
const afterUnloadA = await statusRoute.handler({} as never, ctxA) as { status: number; body: unknown };
|
||||
const afterUnloadB = await statusRoute.handler({} as never, ctxB) as { status: number; body: unknown };
|
||||
expect(afterUnloadA.status).toBe(503);
|
||||
expect(afterUnloadB.status).toBe(200);
|
||||
expect((afterUnloadB.body as { jid: string }).jid).toBe("/repo-b");
|
||||
|
||||
await plugin.hooks!.onUnload!(ctxB);
|
||||
expect(connectionInstances[1]?.stop).toHaveBeenCalledTimes(1);
|
||||
|
||||
const finalStatusA = await statusRoute.handler({} as never, ctxA) as { status: number; body: unknown };
|
||||
const finalStatusB = await statusRoute.handler({} as never, ctxB) as { status: number; body: unknown };
|
||||
expect(finalStatusA.status).toBe(503);
|
||||
expect(finalStatusB.status).toBe(503);
|
||||
});
|
||||
});
|
||||
|
||||
describe("markProcessed retention", () => {
|
||||
it("prunes rows older than retention and keeps recent rows", () => {
|
||||
const db = createInMemoryDb();
|
||||
|
||||
@@ -50,6 +50,10 @@ const settingsSchema: Record<string, PluginSettingSchema> = {
|
||||
|
||||
const connections = new Map<string, WhatsAppConnection>();
|
||||
|
||||
function getConnectionKey(ctx: PluginContext): string {
|
||||
return `${ctx.taskStore.getRootDir()}::${ctx.pluginId}`;
|
||||
}
|
||||
|
||||
export function getSettingString(settings: Record<string, unknown>, key: string): string | undefined {
|
||||
const value = settings[key];
|
||||
return typeof value === "string" && value.trim() ? value.trim() : undefined;
|
||||
@@ -159,7 +163,7 @@ function getDbFromTaskStore(ctx: PluginContext): PluginDb {
|
||||
}
|
||||
|
||||
function getConnectionOrResponse(ctx: PluginContext): { connection?: WhatsAppConnection; error?: PluginRouteResponse } {
|
||||
const connection = connections.get(ctx.pluginId);
|
||||
const connection = connections.get(getConnectionKey(ctx));
|
||||
if (!connection) {
|
||||
return { error: { status: 503, body: { error: "WhatsApp connection is not initialized" } } };
|
||||
}
|
||||
@@ -242,14 +246,15 @@ const plugin: FusionPlugin = definePlugin({
|
||||
onLoad: async (ctx) => {
|
||||
const db = getDbFromTaskStore(ctx);
|
||||
const connection = new WhatsAppConnection(ctx, plugin.manifest.version, generateReply, db);
|
||||
connections.set(ctx.pluginId, connection);
|
||||
connections.set(getConnectionKey(ctx), connection);
|
||||
await connection.start();
|
||||
},
|
||||
onUnload: async () => {
|
||||
for (const [pluginId, connection] of connections.entries()) {
|
||||
await connection.stop();
|
||||
connections.delete(pluginId);
|
||||
}
|
||||
onUnload: async (ctx) => {
|
||||
const connectionKey = getConnectionKey(ctx);
|
||||
const connection = connections.get(connectionKey);
|
||||
if (!connection) return;
|
||||
await connection.stop();
|
||||
connections.delete(connectionKey);
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user