File size: 9,431 Bytes
eb3f11e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 | import {
parseStrictNonNegativeInteger,
parseStrictPositiveInteger,
} from "@openclaw/normalization-core/number-coercion";
import type { Command } from "commander";
import { getChannelPlugin } from "../../../channels/plugins/index.js";
import {
CHANNEL_MESSAGE_ACTION_NAMES,
type ChannelMessageActionName,
} from "../../../channels/plugins/types.public.js";
import { resolveMessageSecretScope } from "../../../cli/message-secret-scope.js";
import { parseAccountSelector } from "../../../commands/channels/account-selector.js";
import { messageCommand } from "../../../commands/message.js";
import { getRuntimeConfig } from "../../../config/config.js";
import { danger, setVerbose } from "../../../globals.js";
import { formatErrorMessage } from "../../../infra/errors.js";
import { CHANNEL_TARGET_DESCRIPTION } from "../../../infra/outbound/channel-target.js";
import { resolveMessageActionOutcome } from "../../../infra/outbound/message-action-contracts.js";
import { createSubsystemLogger } from "../../../logging/subsystem.js";
import { withActivatedPluginIds } from "../../../plugins/activation-context.js";
import {
resolveConfiguredChannelPluginIds,
resolveDiscoverableScopedChannelPluginIds,
} from "../../../plugins/channel-plugin-ids.js";
import { createHookRunner } from "../../../plugins/hooks.js";
import { loadPluginRegistryHandle } from "../../../plugins/loader.js";
import type { PluginRegistry } from "../../../plugins/registry-types.js";
import { withPluginRuntimeRegistryScope } from "../../../plugins/runtime/gateway-request-scope.js";
import { defaultRuntime } from "../../../runtime.js";
import {
ABSOLUTE_DEADLINE_EXPIRED,
awaitWithinDeadline,
} from "../../../utils/absolute-deadline.js";
import { runCommandWithRuntime } from "../../cli-utils.js";
import { createDefaultDeps } from "../../deps.js";
import { requestExitAfterOneShotOutput } from "../../one-shot-exit.js";
/** Shared helpers used by every message subcommand registration. */
export type MessageCliHelpers = {
withMessageBase: (command: Command) => Command;
withMessageTarget: (command: Command) => Command;
withRequiredMessageTarget: (command: Command) => Command;
runMessageAction: (action: string, opts: Record<string, unknown>) => Promise<void>;
};
const GATEWAY_STOP_TIMEOUT_MS = 2500;
const ACTIONS_REQUIRING_CONFIGURED_CHANNEL_PRELOAD = new Set(["broadcast"]);
const CHANNEL_MESSAGE_ACTION_NAME_SET = new Set<string>(CHANNEL_MESSAGE_ACTION_NAMES);
const STRICT_POSITIVE_INTEGER_OPTIONS = new Map([
["pollDurationHours", "--poll-duration-hours"],
["pollDurationSeconds", "--poll-duration-seconds"],
["limit", "--limit"],
["autoArchiveMin", "--auto-archive-min"],
]);
const STRICT_NON_NEGATIVE_INTEGER_OPTIONS = new Map([
["durationMin", "--duration-min"],
["deleteDays", "--delete-days"],
]);
type MessagePluginPreloadPlan = { preload: true; channelId?: string } | { preload: false };
function normalizeMessageOptions(opts: Record<string, unknown>): Record<string, unknown> {
const { account, ...rest } = opts;
return {
...rest,
accountId: typeof account === "string" ? account : rest.accountId,
};
}
function parseMessageChannelSelector(channel: string): string {
if (!channel.trim()) {
throw new Error("--channel must not be blank");
}
return channel;
}
function validateMessageNumericOptions(opts: Record<string, unknown>): void {
for (const [key, flag] of STRICT_POSITIVE_INTEGER_OPTIONS) {
if (opts[key] === undefined) {
continue;
}
if (parseStrictPositiveInteger(opts[key]) === undefined) {
throw new Error(`${flag} must be a positive integer.`);
}
}
for (const [key, flag] of STRICT_NON_NEGATIVE_INTEGER_OPTIONS) {
if (opts[key] === undefined) {
continue;
}
if (parseStrictNonNegativeInteger(opts[key]) === undefined) {
throw new Error(`${flag} must be a non-negative integer.`);
}
}
}
async function runPluginStopHooks(registry: PluginRegistry): Promise<void> {
const runner = createHookRunner(registry, { logger: createSubsystemLogger("plugins") });
const result = await awaitWithinDeadline(
() =>
withPluginRuntimeRegistryScope(registry, () =>
runner.runGatewayStop({ reason: "cli message action complete" }, {}),
),
Date.now() + GATEWAY_STOP_TIMEOUT_MS,
);
if (result === ABSOLUTE_DEADLINE_EXPIRED) {
defaultRuntime.error(
danger(`gateway_stop hook exceeded ${GATEWAY_STOP_TIMEOUT_MS}ms; continuing`),
);
}
}
function resolveScopedMessageChannel(opts: Record<string, unknown>): string | undefined {
return resolveMessageSecretScope({
channel: opts.channel,
target: opts.target,
targets: opts.targets,
}).channel;
}
function asChannelMessageActionName(action: string): ChannelMessageActionName | undefined {
return CHANNEL_MESSAGE_ACTION_NAME_SET.has(action)
? (action as ChannelMessageActionName)
: undefined;
}
function isGatewayOwnedMessageAction(action: string, scopedChannel: string | undefined): boolean {
const messageAction = asChannelMessageActionName(action);
if (!messageAction || !scopedChannel) {
return false;
}
const plugin = getChannelPlugin(scopedChannel);
const executionMode = plugin?.actions?.resolveExecutionMode?.({
action: messageAction,
});
return executionMode === "gateway";
}
function resolveMessagePluginPreloadPlan(
action: string,
opts: Record<string, unknown>,
): MessagePluginPreloadPlan {
const scopedChannel = resolveScopedMessageChannel(opts);
// Gateway-owned actions can execute without loading channel plugins in the CLI process;
// dry-runs, broadcasts, and local actions need registry metadata before building payloads.
if (
opts.dryRun === true ||
ACTIONS_REQUIRING_CONFIGURED_CHANNEL_PRELOAD.has(action) ||
!isGatewayOwnedMessageAction(action, scopedChannel)
) {
return { preload: true, ...(scopedChannel ? { channelId: scopedChannel } : {}) };
}
return { preload: false };
}
/** Create shared option decorators and the common message action runner. */
export function createMessageCliHelpers(messageChannelOptions: string): MessageCliHelpers {
return {
withMessageBase: (command) =>
command
.option(
"--channel <channel>",
`Channel: ${messageChannelOptions}`,
parseMessageChannelSelector,
)
.option("--account <id>", "Channel account id (accountId)", parseAccountSelector)
.option("--json", "Output result as JSON", false)
.option("--dry-run", "Print payload and skip sending", false)
.option("--verbose", "Verbose logging", false),
withMessageTarget: (command) =>
command.option("-t, --target <dest>", CHANNEL_TARGET_DESCRIPTION),
withRequiredMessageTarget: (command) =>
command.requiredOption("-t, --target <dest>", CHANNEL_TARGET_DESCRIPTION),
runMessageAction: async (action, opts) => {
setVerbose(Boolean(opts.verbose));
let failed = false;
let result: Awaited<ReturnType<typeof messageCommand>> | undefined;
let pluginRegistry: PluginRegistry | undefined;
try {
await runCommandWithRuntime(
defaultRuntime,
async () => {
validateMessageNumericOptions(opts);
if (action === "poll" && opts.pollAnonymous === true && opts.pollPublic === true) {
throw new Error("--poll-anonymous and --poll-public are mutually exclusive.");
}
const preloadPlan = resolveMessagePluginPreloadPlan(action, opts);
if (preloadPlan.preload) {
const config = getRuntimeConfig();
const pluginIds = preloadPlan.channelId
? resolveDiscoverableScopedChannelPluginIds({
config,
activationSourceConfig: config,
channelIds: [preloadPlan.channelId],
env: process.env,
})
: resolveConfiguredChannelPluginIds({
config,
activationSourceConfig: config,
env: process.env,
});
const activatedConfig = withActivatedPluginIds({ config, pluginIds }) ?? config;
pluginRegistry = loadPluginRegistryHandle({
config: activatedConfig,
activationSourceConfig: activatedConfig,
onlyPluginIds: pluginIds,
throwOnLoadError: true,
});
}
const deps = createDefaultDeps();
const run = () =>
messageCommand(
{
...normalizeMessageOptions(opts),
action,
},
deps,
defaultRuntime,
);
result = await withPluginRuntimeRegistryScope(pluginRegistry, run);
},
(err) => {
failed = true;
defaultRuntime.error(danger(formatErrorMessage(err)));
},
);
} finally {
// Finalize only this command's registry, including JSON/expected errors that rethrow.
if (pluginRegistry && action !== "read") {
await runPluginStopHooks(pluginRegistry);
}
}
failed ||= result !== undefined && !resolveMessageActionOutcome(result).ok;
requestExitAfterOneShotOutput(defaultRuntime, failed ? 1 : 0);
},
};
}
|