mirror of
https://github.com/openclaw/openclaw.git
synced 2026-03-12 07:20:45 +00:00
refactor: share passive account lifecycle helpers
This commit is contained in:
@@ -21,6 +21,7 @@ import {
|
|||||||
import {
|
import {
|
||||||
buildAccountScopedDmSecurityPolicy,
|
buildAccountScopedDmSecurityPolicy,
|
||||||
collectOpenGroupPolicyRestrictSendersWarnings,
|
collectOpenGroupPolicyRestrictSendersWarnings,
|
||||||
|
createAccountStatusSink,
|
||||||
formatNormalizedAllowFromEntries,
|
formatNormalizedAllowFromEntries,
|
||||||
mapAllowFromEntries,
|
mapAllowFromEntries,
|
||||||
} from "openclaw/plugin-sdk/compat";
|
} from "openclaw/plugin-sdk/compat";
|
||||||
@@ -369,8 +370,11 @@ export const bluebubblesPlugin: ChannelPlugin<ResolvedBlueBubblesAccount> = {
|
|||||||
startAccount: async (ctx) => {
|
startAccount: async (ctx) => {
|
||||||
const account = ctx.account;
|
const account = ctx.account;
|
||||||
const webhookPath = resolveWebhookPathFromConfig(account.config);
|
const webhookPath = resolveWebhookPathFromConfig(account.config);
|
||||||
ctx.setStatus({
|
const statusSink = createAccountStatusSink({
|
||||||
accountId: account.accountId,
|
accountId: ctx.accountId,
|
||||||
|
setStatus: ctx.setStatus,
|
||||||
|
});
|
||||||
|
statusSink({
|
||||||
baseUrl: account.baseUrl,
|
baseUrl: account.baseUrl,
|
||||||
});
|
});
|
||||||
ctx.log?.info(`[${account.accountId}] starting provider (webhook=${webhookPath})`);
|
ctx.log?.info(`[${account.accountId}] starting provider (webhook=${webhookPath})`);
|
||||||
@@ -379,7 +383,7 @@ export const bluebubblesPlugin: ChannelPlugin<ResolvedBlueBubblesAccount> = {
|
|||||||
config: ctx.cfg,
|
config: ctx.cfg,
|
||||||
runtime: ctx.runtime,
|
runtime: ctx.runtime,
|
||||||
abortSignal: ctx.abortSignal,
|
abortSignal: ctx.abortSignal,
|
||||||
statusSink: (patch) => ctx.setStatus({ accountId: ctx.accountId, ...patch }),
|
statusSink,
|
||||||
webhookPath,
|
webhookPath,
|
||||||
});
|
});
|
||||||
},
|
},
|
||||||
|
|||||||
@@ -12,6 +12,7 @@ import {
|
|||||||
buildComputedAccountStatusSnapshot,
|
buildComputedAccountStatusSnapshot,
|
||||||
buildChannelConfigSchema,
|
buildChannelConfigSchema,
|
||||||
DEFAULT_ACCOUNT_ID,
|
DEFAULT_ACCOUNT_ID,
|
||||||
|
createAccountStatusSink,
|
||||||
getChatChannelMeta,
|
getChatChannelMeta,
|
||||||
listDirectoryGroupEntriesFromMapKeys,
|
listDirectoryGroupEntriesFromMapKeys,
|
||||||
listDirectoryUserEntriesFromAllowFrom,
|
listDirectoryUserEntriesFromAllowFrom,
|
||||||
@@ -21,6 +22,7 @@ import {
|
|||||||
PAIRING_APPROVED_MESSAGE,
|
PAIRING_APPROVED_MESSAGE,
|
||||||
resolveChannelMediaMaxBytes,
|
resolveChannelMediaMaxBytes,
|
||||||
resolveGoogleChatGroupRequireMention,
|
resolveGoogleChatGroupRequireMention,
|
||||||
|
runPassiveAccountLifecycle,
|
||||||
type ChannelDock,
|
type ChannelDock,
|
||||||
type ChannelMessageActionAdapter,
|
type ChannelMessageActionAdapter,
|
||||||
type ChannelPlugin,
|
type ChannelPlugin,
|
||||||
@@ -509,37 +511,39 @@ export const googlechatPlugin: ChannelPlugin<ResolvedGoogleChatAccount> = {
|
|||||||
gateway: {
|
gateway: {
|
||||||
startAccount: async (ctx) => {
|
startAccount: async (ctx) => {
|
||||||
const account = ctx.account;
|
const account = ctx.account;
|
||||||
ctx.log?.info(`[${account.accountId}] starting Google Chat webhook`);
|
const statusSink = createAccountStatusSink({
|
||||||
ctx.setStatus({
|
|
||||||
accountId: account.accountId,
|
accountId: account.accountId,
|
||||||
|
setStatus: ctx.setStatus,
|
||||||
|
});
|
||||||
|
ctx.log?.info(`[${account.accountId}] starting Google Chat webhook`);
|
||||||
|
statusSink({
|
||||||
running: true,
|
running: true,
|
||||||
lastStartAt: Date.now(),
|
lastStartAt: Date.now(),
|
||||||
webhookPath: resolveGoogleChatWebhookPath({ account }),
|
webhookPath: resolveGoogleChatWebhookPath({ account }),
|
||||||
audienceType: account.config.audienceType,
|
audienceType: account.config.audienceType,
|
||||||
audience: account.config.audience,
|
audience: account.config.audience,
|
||||||
});
|
});
|
||||||
const unregister = await startGoogleChatMonitor({
|
await runPassiveAccountLifecycle({
|
||||||
account,
|
|
||||||
config: ctx.cfg,
|
|
||||||
runtime: ctx.runtime,
|
|
||||||
abortSignal: ctx.abortSignal,
|
abortSignal: ctx.abortSignal,
|
||||||
webhookPath: account.config.webhookPath,
|
start: async () =>
|
||||||
webhookUrl: account.config.webhookUrl,
|
await startGoogleChatMonitor({
|
||||||
statusSink: (patch) => ctx.setStatus({ accountId: account.accountId, ...patch }),
|
account,
|
||||||
});
|
config: ctx.cfg,
|
||||||
// Keep the promise pending until abort (webhook mode is passive).
|
runtime: ctx.runtime,
|
||||||
await new Promise<void>((resolve) => {
|
abortSignal: ctx.abortSignal,
|
||||||
if (ctx.abortSignal.aborted) {
|
webhookPath: account.config.webhookPath,
|
||||||
resolve();
|
webhookUrl: account.config.webhookUrl,
|
||||||
return;
|
statusSink,
|
||||||
}
|
}),
|
||||||
ctx.abortSignal.addEventListener("abort", () => resolve(), { once: true });
|
stop: async (unregister) => {
|
||||||
});
|
unregister?.();
|
||||||
unregister?.();
|
},
|
||||||
ctx.setStatus({
|
onStop: async () => {
|
||||||
accountId: account.accountId,
|
statusSink({
|
||||||
running: false,
|
running: false,
|
||||||
lastStopAt: Date.now(),
|
lastStopAt: Date.now(),
|
||||||
|
});
|
||||||
|
},
|
||||||
});
|
});
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
|
|||||||
67
extensions/irc/src/channel.startup.test.ts
Normal file
67
extensions/irc/src/channel.startup.test.ts
Normal file
@@ -0,0 +1,67 @@
|
|||||||
|
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||||
|
import { createStartAccountContext } from "../../test-utils/start-account-context.js";
|
||||||
|
import type { ResolvedIrcAccount } from "./accounts.js";
|
||||||
|
|
||||||
|
const hoisted = vi.hoisted(() => ({
|
||||||
|
monitorIrcProvider: vi.fn(),
|
||||||
|
}));
|
||||||
|
|
||||||
|
vi.mock("./monitor.js", async () => {
|
||||||
|
const actual = await vi.importActual<typeof import("./monitor.js")>("./monitor.js");
|
||||||
|
return {
|
||||||
|
...actual,
|
||||||
|
monitorIrcProvider: hoisted.monitorIrcProvider,
|
||||||
|
};
|
||||||
|
});
|
||||||
|
|
||||||
|
import { ircPlugin } from "./channel.js";
|
||||||
|
|
||||||
|
describe("ircPlugin gateway.startAccount", () => {
|
||||||
|
afterEach(() => {
|
||||||
|
vi.clearAllMocks();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("keeps startAccount pending until abort, then stops the monitor", async () => {
|
||||||
|
const stop = vi.fn();
|
||||||
|
hoisted.monitorIrcProvider.mockResolvedValue({ stop });
|
||||||
|
|
||||||
|
const account: ResolvedIrcAccount = {
|
||||||
|
accountId: "default",
|
||||||
|
enabled: true,
|
||||||
|
name: "default",
|
||||||
|
configured: true,
|
||||||
|
host: "irc.example.com",
|
||||||
|
port: 6697,
|
||||||
|
tls: true,
|
||||||
|
nick: "openclaw",
|
||||||
|
username: "openclaw",
|
||||||
|
realname: "OpenClaw",
|
||||||
|
password: "",
|
||||||
|
passwordSource: "none",
|
||||||
|
config: {} as ResolvedIrcAccount["config"],
|
||||||
|
};
|
||||||
|
|
||||||
|
const abort = new AbortController();
|
||||||
|
const task = ircPlugin.gateway!.startAccount!(
|
||||||
|
createStartAccountContext({
|
||||||
|
account,
|
||||||
|
abortSignal: abort.signal,
|
||||||
|
}),
|
||||||
|
);
|
||||||
|
let settled = false;
|
||||||
|
void task.then(() => {
|
||||||
|
settled = true;
|
||||||
|
});
|
||||||
|
|
||||||
|
await vi.waitFor(() => {
|
||||||
|
expect(hoisted.monitorIrcProvider).toHaveBeenCalledOnce();
|
||||||
|
});
|
||||||
|
expect(settled).toBe(false);
|
||||||
|
expect(stop).not.toHaveBeenCalled();
|
||||||
|
|
||||||
|
abort.abort();
|
||||||
|
await task;
|
||||||
|
|
||||||
|
expect(stop).toHaveBeenCalledOnce();
|
||||||
|
});
|
||||||
|
});
|
||||||
@@ -9,10 +9,12 @@ import {
|
|||||||
buildBaseAccountStatusSnapshot,
|
buildBaseAccountStatusSnapshot,
|
||||||
buildBaseChannelStatusSummary,
|
buildBaseChannelStatusSummary,
|
||||||
buildChannelConfigSchema,
|
buildChannelConfigSchema,
|
||||||
|
createAccountStatusSink,
|
||||||
DEFAULT_ACCOUNT_ID,
|
DEFAULT_ACCOUNT_ID,
|
||||||
deleteAccountFromConfigSection,
|
deleteAccountFromConfigSection,
|
||||||
getChatChannelMeta,
|
getChatChannelMeta,
|
||||||
PAIRING_APPROVED_MESSAGE,
|
PAIRING_APPROVED_MESSAGE,
|
||||||
|
runPassiveAccountLifecycle,
|
||||||
setAccountEnabledInConfigSection,
|
setAccountEnabledInConfigSection,
|
||||||
type ChannelPlugin,
|
type ChannelPlugin,
|
||||||
} from "openclaw/plugin-sdk/irc";
|
} from "openclaw/plugin-sdk/irc";
|
||||||
@@ -353,6 +355,10 @@ export const ircPlugin: ChannelPlugin<ResolvedIrcAccount, IrcProbe> = {
|
|||||||
gateway: {
|
gateway: {
|
||||||
startAccount: async (ctx) => {
|
startAccount: async (ctx) => {
|
||||||
const account = ctx.account;
|
const account = ctx.account;
|
||||||
|
const statusSink = createAccountStatusSink({
|
||||||
|
accountId: ctx.accountId,
|
||||||
|
setStatus: ctx.setStatus,
|
||||||
|
});
|
||||||
if (!account.configured) {
|
if (!account.configured) {
|
||||||
throw new Error(
|
throw new Error(
|
||||||
`IRC is not configured for account "${account.accountId}" (need host and nick in channels.irc).`,
|
`IRC is not configured for account "${account.accountId}" (need host and nick in channels.irc).`,
|
||||||
@@ -361,14 +367,20 @@ export const ircPlugin: ChannelPlugin<ResolvedIrcAccount, IrcProbe> = {
|
|||||||
ctx.log?.info(
|
ctx.log?.info(
|
||||||
`[${account.accountId}] starting IRC provider (${account.host}:${account.port}${account.tls ? " tls" : ""})`,
|
`[${account.accountId}] starting IRC provider (${account.host}:${account.port}${account.tls ? " tls" : ""})`,
|
||||||
);
|
);
|
||||||
const { stop } = await monitorIrcProvider({
|
await runPassiveAccountLifecycle({
|
||||||
accountId: account.accountId,
|
|
||||||
config: ctx.cfg as CoreConfig,
|
|
||||||
runtime: ctx.runtime,
|
|
||||||
abortSignal: ctx.abortSignal,
|
abortSignal: ctx.abortSignal,
|
||||||
statusSink: (patch) => ctx.setStatus({ accountId: ctx.accountId, ...patch }),
|
start: async () =>
|
||||||
|
await monitorIrcProvider({
|
||||||
|
accountId: account.accountId,
|
||||||
|
config: ctx.cfg as CoreConfig,
|
||||||
|
runtime: ctx.runtime,
|
||||||
|
abortSignal: ctx.abortSignal,
|
||||||
|
statusSink,
|
||||||
|
}),
|
||||||
|
stop: async (monitor) => {
|
||||||
|
monitor.stop();
|
||||||
|
},
|
||||||
});
|
});
|
||||||
return { stop };
|
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
};
|
};
|
||||||
|
|||||||
@@ -9,6 +9,7 @@ import {
|
|||||||
applySetupAccountConfigPatch,
|
applySetupAccountConfigPatch,
|
||||||
buildComputedAccountStatusSnapshot,
|
buildComputedAccountStatusSnapshot,
|
||||||
buildChannelConfigSchema,
|
buildChannelConfigSchema,
|
||||||
|
createAccountStatusSink,
|
||||||
DEFAULT_ACCOUNT_ID,
|
DEFAULT_ACCOUNT_ID,
|
||||||
deleteAccountFromConfigSection,
|
deleteAccountFromConfigSection,
|
||||||
migrateBaseNameToDefaultAccount,
|
migrateBaseNameToDefaultAccount,
|
||||||
@@ -500,8 +501,11 @@ export const mattermostPlugin: ChannelPlugin<ResolvedMattermostAccount> = {
|
|||||||
gateway: {
|
gateway: {
|
||||||
startAccount: async (ctx) => {
|
startAccount: async (ctx) => {
|
||||||
const account = ctx.account;
|
const account = ctx.account;
|
||||||
ctx.setStatus({
|
const statusSink = createAccountStatusSink({
|
||||||
accountId: account.accountId,
|
accountId: ctx.accountId,
|
||||||
|
setStatus: ctx.setStatus,
|
||||||
|
});
|
||||||
|
statusSink({
|
||||||
baseUrl: account.baseUrl,
|
baseUrl: account.baseUrl,
|
||||||
botTokenSource: account.botTokenSource,
|
botTokenSource: account.botTokenSource,
|
||||||
});
|
});
|
||||||
@@ -513,7 +517,7 @@ export const mattermostPlugin: ChannelPlugin<ResolvedMattermostAccount> = {
|
|||||||
config: ctx.cfg,
|
config: ctx.cfg,
|
||||||
runtime: ctx.runtime,
|
runtime: ctx.runtime,
|
||||||
abortSignal: ctx.abortSignal,
|
abortSignal: ctx.abortSignal,
|
||||||
statusSink: (patch) => ctx.setStatus({ accountId: ctx.accountId, ...patch }),
|
statusSink,
|
||||||
});
|
});
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
|
|||||||
@@ -2,8 +2,10 @@ import {
|
|||||||
buildAccountScopedDmSecurityPolicy,
|
buildAccountScopedDmSecurityPolicy,
|
||||||
collectAllowlistProviderGroupPolicyWarnings,
|
collectAllowlistProviderGroupPolicyWarnings,
|
||||||
collectOpenGroupPolicyRouteAllowlistWarnings,
|
collectOpenGroupPolicyRouteAllowlistWarnings,
|
||||||
|
createAccountStatusSink,
|
||||||
formatAllowFromLowercase,
|
formatAllowFromLowercase,
|
||||||
mapAllowFromEntries,
|
mapAllowFromEntries,
|
||||||
|
runPassiveAccountLifecycle,
|
||||||
} from "openclaw/plugin-sdk/compat";
|
} from "openclaw/plugin-sdk/compat";
|
||||||
import {
|
import {
|
||||||
applyAccountNameToChannelSection,
|
applyAccountNameToChannelSection,
|
||||||
@@ -15,7 +17,6 @@ import {
|
|||||||
deleteAccountFromConfigSection,
|
deleteAccountFromConfigSection,
|
||||||
normalizeAccountId,
|
normalizeAccountId,
|
||||||
setAccountEnabledInConfigSection,
|
setAccountEnabledInConfigSection,
|
||||||
waitForAbortSignal,
|
|
||||||
type ChannelPlugin,
|
type ChannelPlugin,
|
||||||
type OpenClawConfig,
|
type OpenClawConfig,
|
||||||
type ChannelSetupInput,
|
type ChannelSetupInput,
|
||||||
@@ -338,17 +339,25 @@ export const nextcloudTalkPlugin: ChannelPlugin<ResolvedNextcloudTalkAccount> =
|
|||||||
|
|
||||||
ctx.log?.info(`[${account.accountId}] starting Nextcloud Talk webhook server`);
|
ctx.log?.info(`[${account.accountId}] starting Nextcloud Talk webhook server`);
|
||||||
|
|
||||||
const { stop } = await monitorNextcloudTalkProvider({
|
const statusSink = createAccountStatusSink({
|
||||||
accountId: account.accountId,
|
accountId: ctx.accountId,
|
||||||
config: ctx.cfg as CoreConfig,
|
setStatus: ctx.setStatus,
|
||||||
runtime: ctx.runtime,
|
|
||||||
abortSignal: ctx.abortSignal,
|
|
||||||
statusSink: (patch) => ctx.setStatus({ accountId: ctx.accountId, ...patch }),
|
|
||||||
});
|
});
|
||||||
|
|
||||||
// Keep webhook channels pending for the account lifecycle.
|
await runPassiveAccountLifecycle({
|
||||||
await waitForAbortSignal(ctx.abortSignal);
|
abortSignal: ctx.abortSignal,
|
||||||
stop();
|
start: async () =>
|
||||||
|
await monitorNextcloudTalkProvider({
|
||||||
|
accountId: account.accountId,
|
||||||
|
config: ctx.cfg as CoreConfig,
|
||||||
|
runtime: ctx.runtime,
|
||||||
|
abortSignal: ctx.abortSignal,
|
||||||
|
statusSink,
|
||||||
|
}),
|
||||||
|
stop: async (monitor) => {
|
||||||
|
monitor.stop();
|
||||||
|
},
|
||||||
|
});
|
||||||
},
|
},
|
||||||
logoutAccount: async ({ accountId, cfg }) => {
|
logoutAccount: async ({ accountId, cfg }) => {
|
||||||
const nextCfg = { ...cfg } as OpenClawConfig;
|
const nextCfg = { ...cfg } as OpenClawConfig;
|
||||||
|
|||||||
@@ -1,8 +1,9 @@
|
|||||||
import {
|
import {
|
||||||
buildAccountScopedDmSecurityPolicy,
|
buildAccountScopedDmSecurityPolicy,
|
||||||
collectOpenProviderGroupPolicyWarnings,
|
|
||||||
buildOpenGroupPolicyRestrictSendersWarning,
|
buildOpenGroupPolicyRestrictSendersWarning,
|
||||||
buildOpenGroupPolicyWarning,
|
buildOpenGroupPolicyWarning,
|
||||||
|
collectOpenProviderGroupPolicyWarnings,
|
||||||
|
createAccountStatusSink,
|
||||||
mapAllowFromEntries,
|
mapAllowFromEntries,
|
||||||
} from "openclaw/plugin-sdk/compat";
|
} from "openclaw/plugin-sdk/compat";
|
||||||
import type {
|
import type {
|
||||||
@@ -357,6 +358,10 @@ export const zaloPlugin: ChannelPlugin<ResolvedZaloAccount> = {
|
|||||||
`[${account.accountId}] Zalo probe threw before provider start: ${err instanceof Error ? (err.stack ?? err.message) : String(err)}`,
|
`[${account.accountId}] Zalo probe threw before provider start: ${err instanceof Error ? (err.stack ?? err.message) : String(err)}`,
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
const statusSink = createAccountStatusSink({
|
||||||
|
accountId: ctx.accountId,
|
||||||
|
setStatus: ctx.setStatus,
|
||||||
|
});
|
||||||
ctx.log?.info(`[${account.accountId}] starting provider${zaloBotLabel} mode=${mode}`);
|
ctx.log?.info(`[${account.accountId}] starting provider${zaloBotLabel} mode=${mode}`);
|
||||||
const { monitorZaloProvider } = await import("./monitor.js");
|
const { monitorZaloProvider } = await import("./monitor.js");
|
||||||
return monitorZaloProvider({
|
return monitorZaloProvider({
|
||||||
@@ -370,7 +375,7 @@ export const zaloPlugin: ChannelPlugin<ResolvedZaloAccount> = {
|
|||||||
webhookSecret: normalizeSecretInputString(account.config.webhookSecret),
|
webhookSecret: normalizeSecretInputString(account.config.webhookSecret),
|
||||||
webhookPath: account.config.webhookPath,
|
webhookPath: account.config.webhookPath,
|
||||||
fetcher,
|
fetcher,
|
||||||
statusSink: (patch) => ctx.setStatus({ accountId: ctx.accountId, ...patch }),
|
statusSink,
|
||||||
});
|
});
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
|
|||||||
@@ -1,5 +1,6 @@
|
|||||||
import {
|
import {
|
||||||
buildAccountScopedDmSecurityPolicy,
|
buildAccountScopedDmSecurityPolicy,
|
||||||
|
createAccountStatusSink,
|
||||||
mapAllowFromEntries,
|
mapAllowFromEntries,
|
||||||
} from "openclaw/plugin-sdk/compat";
|
} from "openclaw/plugin-sdk/compat";
|
||||||
import type {
|
import type {
|
||||||
@@ -682,6 +683,10 @@ export const zalouserPlugin: ChannelPlugin<ResolvedZalouserAccount> = {
|
|||||||
} catch {
|
} catch {
|
||||||
// ignore probe errors
|
// ignore probe errors
|
||||||
}
|
}
|
||||||
|
const statusSink = createAccountStatusSink({
|
||||||
|
accountId: ctx.accountId,
|
||||||
|
setStatus: ctx.setStatus,
|
||||||
|
});
|
||||||
ctx.log?.info(`[${account.accountId}] starting zalouser provider${userLabel}`);
|
ctx.log?.info(`[${account.accountId}] starting zalouser provider${userLabel}`);
|
||||||
const { monitorZalouserProvider } = await import("./monitor.js");
|
const { monitorZalouserProvider } = await import("./monitor.js");
|
||||||
return monitorZalouserProvider({
|
return monitorZalouserProvider({
|
||||||
@@ -689,7 +694,7 @@ export const zalouserPlugin: ChannelPlugin<ResolvedZalouserAccount> = {
|
|||||||
config: ctx.cfg,
|
config: ctx.cfg,
|
||||||
runtime: ctx.runtime,
|
runtime: ctx.runtime,
|
||||||
abortSignal: ctx.abortSignal,
|
abortSignal: ctx.abortSignal,
|
||||||
statusSink: (patch) => ctx.setStatus({ accountId: ctx.accountId, ...patch }),
|
statusSink,
|
||||||
});
|
});
|
||||||
},
|
},
|
||||||
loginWithQrStart: async (params) => {
|
loginWithQrStart: async (params) => {
|
||||||
|
|||||||
@@ -1,6 +1,11 @@
|
|||||||
import { EventEmitter } from "node:events";
|
import { EventEmitter } from "node:events";
|
||||||
import { describe, expect, it, vi } from "vitest";
|
import { describe, expect, it, vi } from "vitest";
|
||||||
import { keepHttpServerTaskAlive, waitUntilAbort } from "./channel-lifecycle.js";
|
import {
|
||||||
|
createAccountStatusSink,
|
||||||
|
keepHttpServerTaskAlive,
|
||||||
|
runPassiveAccountLifecycle,
|
||||||
|
waitUntilAbort,
|
||||||
|
} from "./channel-lifecycle.js";
|
||||||
|
|
||||||
type FakeServer = EventEmitter & {
|
type FakeServer = EventEmitter & {
|
||||||
close: (callback?: () => void) => void;
|
close: (callback?: () => void) => void;
|
||||||
@@ -18,6 +23,22 @@ function createFakeServer(): FakeServer {
|
|||||||
}
|
}
|
||||||
|
|
||||||
describe("plugin-sdk channel lifecycle helpers", () => {
|
describe("plugin-sdk channel lifecycle helpers", () => {
|
||||||
|
it("binds account id onto status patches", () => {
|
||||||
|
const setStatus = vi.fn();
|
||||||
|
const statusSink = createAccountStatusSink({
|
||||||
|
accountId: "default",
|
||||||
|
setStatus,
|
||||||
|
});
|
||||||
|
|
||||||
|
statusSink({ running: true, lastStartAt: 123 });
|
||||||
|
|
||||||
|
expect(setStatus).toHaveBeenCalledWith({
|
||||||
|
accountId: "default",
|
||||||
|
running: true,
|
||||||
|
lastStartAt: 123,
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
it("resolves waitUntilAbort when signal aborts", async () => {
|
it("resolves waitUntilAbort when signal aborts", async () => {
|
||||||
const abort = new AbortController();
|
const abort = new AbortController();
|
||||||
const task = waitUntilAbort(abort.signal);
|
const task = waitUntilAbort(abort.signal);
|
||||||
@@ -32,6 +53,40 @@ describe("plugin-sdk channel lifecycle helpers", () => {
|
|||||||
await expect(task).resolves.toBeUndefined();
|
await expect(task).resolves.toBeUndefined();
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("runs abort cleanup before resolving", async () => {
|
||||||
|
const abort = new AbortController();
|
||||||
|
const onAbort = vi.fn(async () => undefined);
|
||||||
|
|
||||||
|
const task = waitUntilAbort(abort.signal, onAbort);
|
||||||
|
abort.abort();
|
||||||
|
|
||||||
|
await expect(task).resolves.toBeUndefined();
|
||||||
|
expect(onAbort).toHaveBeenCalledOnce();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("keeps passive account lifecycle pending until abort, then stops once", async () => {
|
||||||
|
const abort = new AbortController();
|
||||||
|
const stop = vi.fn();
|
||||||
|
const task = runPassiveAccountLifecycle({
|
||||||
|
abortSignal: abort.signal,
|
||||||
|
start: async () => ({ stop }),
|
||||||
|
stop: async (handle) => {
|
||||||
|
handle.stop();
|
||||||
|
},
|
||||||
|
});
|
||||||
|
|
||||||
|
const early = await Promise.race([
|
||||||
|
task.then(() => "resolved"),
|
||||||
|
new Promise<"pending">((resolve) => setTimeout(() => resolve("pending"), 25)),
|
||||||
|
]);
|
||||||
|
expect(early).toBe("pending");
|
||||||
|
expect(stop).not.toHaveBeenCalled();
|
||||||
|
|
||||||
|
abort.abort();
|
||||||
|
await expect(task).resolves.toBeUndefined();
|
||||||
|
expect(stop).toHaveBeenCalledOnce();
|
||||||
|
});
|
||||||
|
|
||||||
it("keeps server task pending until close, then resolves", async () => {
|
it("keeps server task pending until close, then resolves", async () => {
|
||||||
const server = createFakeServer();
|
const server = createFakeServer();
|
||||||
const task = keepHttpServerTaskAlive({ server });
|
const task = keepHttpServerTaskAlive({ server });
|
||||||
|
|||||||
@@ -1,25 +1,66 @@
|
|||||||
|
import type { ChannelAccountSnapshot } from "../channels/plugins/types.core.js";
|
||||||
|
|
||||||
type CloseAwareServer = {
|
type CloseAwareServer = {
|
||||||
once: (event: "close", listener: () => void) => unknown;
|
once: (event: "close", listener: () => void) => unknown;
|
||||||
};
|
};
|
||||||
|
|
||||||
|
type PassiveAccountLifecycleParams<Handle> = {
|
||||||
|
abortSignal?: AbortSignal;
|
||||||
|
start: () => Promise<Handle>;
|
||||||
|
stop?: (handle: Handle) => void | Promise<void>;
|
||||||
|
onStop?: () => void | Promise<void>;
|
||||||
|
};
|
||||||
|
|
||||||
|
export function createAccountStatusSink(params: {
|
||||||
|
accountId: string;
|
||||||
|
setStatus: (next: ChannelAccountSnapshot) => void;
|
||||||
|
}): (patch: Omit<ChannelAccountSnapshot, "accountId">) => void {
|
||||||
|
return (patch) => {
|
||||||
|
params.setStatus({ accountId: params.accountId, ...patch });
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Return a promise that resolves when the signal is aborted.
|
* Return a promise that resolves when the signal is aborted.
|
||||||
*
|
*
|
||||||
* If no signal is provided, the promise stays pending forever.
|
* If no signal is provided, the promise stays pending forever. When provided,
|
||||||
|
* `onAbort` runs once before the promise resolves.
|
||||||
*/
|
*/
|
||||||
export function waitUntilAbort(signal?: AbortSignal): Promise<void> {
|
export function waitUntilAbort(
|
||||||
return new Promise<void>((resolve) => {
|
signal?: AbortSignal,
|
||||||
|
onAbort?: () => void | Promise<void>,
|
||||||
|
): Promise<void> {
|
||||||
|
return new Promise<void>((resolve, reject) => {
|
||||||
|
const complete = () => {
|
||||||
|
Promise.resolve(onAbort?.()).then(() => resolve(), reject);
|
||||||
|
};
|
||||||
if (!signal) {
|
if (!signal) {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
if (signal.aborted) {
|
if (signal.aborted) {
|
||||||
resolve();
|
complete();
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
signal.addEventListener("abort", () => resolve(), { once: true });
|
signal.addEventListener("abort", complete, { once: true });
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Keep a passive account task alive until abort, then run optional cleanup.
|
||||||
|
*/
|
||||||
|
export async function runPassiveAccountLifecycle<Handle>(
|
||||||
|
params: PassiveAccountLifecycleParams<Handle>,
|
||||||
|
): Promise<void> {
|
||||||
|
const handle = await params.start();
|
||||||
|
|
||||||
|
try {
|
||||||
|
await waitUntilAbort(params.abortSignal);
|
||||||
|
} finally {
|
||||||
|
await params.stop?.(handle);
|
||||||
|
await params.onStop?.();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Keep a channel/provider task pending until the HTTP server closes.
|
* Keep a channel/provider task pending until the HTTP server closes.
|
||||||
*
|
*
|
||||||
|
|||||||
@@ -20,6 +20,7 @@ export {
|
|||||||
} from "../channels/plugins/directory-config-helpers.js";
|
} from "../channels/plugins/directory-config-helpers.js";
|
||||||
export { buildComputedAccountStatusSnapshot } from "./status-helpers.js";
|
export { buildComputedAccountStatusSnapshot } from "./status-helpers.js";
|
||||||
export { buildChannelConfigSchema } from "../channels/plugins/config-schema.js";
|
export { buildChannelConfigSchema } from "../channels/plugins/config-schema.js";
|
||||||
|
export { createAccountStatusSink, runPassiveAccountLifecycle } from "./channel-lifecycle.js";
|
||||||
export { resolveGoogleChatGroupRequireMention } from "../channels/plugins/group-mentions.js";
|
export { resolveGoogleChatGroupRequireMention } from "../channels/plugins/group-mentions.js";
|
||||||
export { formatPairingApproveHint } from "../channels/plugins/helpers.js";
|
export { formatPairingApproveHint } from "../channels/plugins/helpers.js";
|
||||||
export { resolveChannelMediaMaxBytes } from "../channels/plugins/media-limits.js";
|
export { resolveChannelMediaMaxBytes } from "../channels/plugins/media-limits.js";
|
||||||
|
|||||||
@@ -173,7 +173,12 @@ export {
|
|||||||
WEBHOOK_IN_FLIGHT_DEFAULTS,
|
WEBHOOK_IN_FLIGHT_DEFAULTS,
|
||||||
} from "./webhook-request-guards.js";
|
} from "./webhook-request-guards.js";
|
||||||
export type { WebhookBodyReadProfile, WebhookInFlightLimiter } from "./webhook-request-guards.js";
|
export type { WebhookBodyReadProfile, WebhookInFlightLimiter } from "./webhook-request-guards.js";
|
||||||
export { keepHttpServerTaskAlive, waitUntilAbort } from "./channel-lifecycle.js";
|
export {
|
||||||
|
createAccountStatusSink,
|
||||||
|
keepHttpServerTaskAlive,
|
||||||
|
runPassiveAccountLifecycle,
|
||||||
|
waitUntilAbort,
|
||||||
|
} from "./channel-lifecycle.js";
|
||||||
export type { AgentMediaPayload } from "./agent-media-payload.js";
|
export type { AgentMediaPayload } from "./agent-media-payload.js";
|
||||||
export { buildAgentMediaPayload } from "./agent-media-payload.js";
|
export { buildAgentMediaPayload } from "./agent-media-payload.js";
|
||||||
export {
|
export {
|
||||||
|
|||||||
@@ -61,6 +61,7 @@ export type { PluginRuntime } from "../plugins/runtime/types.js";
|
|||||||
export type { OpenClawPluginApi } from "../plugins/types.js";
|
export type { OpenClawPluginApi } from "../plugins/types.js";
|
||||||
export { DEFAULT_ACCOUNT_ID } from "../routing/session-key.js";
|
export { DEFAULT_ACCOUNT_ID } from "../routing/session-key.js";
|
||||||
export type { RuntimeEnv } from "../runtime.js";
|
export type { RuntimeEnv } from "../runtime.js";
|
||||||
|
export { createAccountStatusSink, runPassiveAccountLifecycle } from "./channel-lifecycle.js";
|
||||||
export {
|
export {
|
||||||
readStoreAllowFromForDmPolicy,
|
readStoreAllowFromForDmPolicy,
|
||||||
resolveEffectiveAllowFromLists,
|
resolveEffectiveAllowFromLists,
|
||||||
|
|||||||
@@ -41,6 +41,7 @@ export {
|
|||||||
applySetupAccountConfigPatch,
|
applySetupAccountConfigPatch,
|
||||||
migrateBaseNameToDefaultAccount,
|
migrateBaseNameToDefaultAccount,
|
||||||
} from "../channels/plugins/setup-helpers.js";
|
} from "../channels/plugins/setup-helpers.js";
|
||||||
|
export { createAccountStatusSink } from "./channel-lifecycle.js";
|
||||||
export { buildComputedAccountStatusSnapshot } from "./status-helpers.js";
|
export { buildComputedAccountStatusSnapshot } from "./status-helpers.js";
|
||||||
export { createAccountListHelpers } from "../channels/plugins/account-helpers.js";
|
export { createAccountListHelpers } from "../channels/plugins/account-helpers.js";
|
||||||
export type {
|
export type {
|
||||||
|
|||||||
Reference in New Issue
Block a user