feat: wire rule registry and client-authenticated callback into server runtime
- Add ruleRegistry and onClientAuthenticated options to YonexusServerRuntime - Dispatch rewritten rule messages (rule::sender::content) to rule registry - Guard onClientAuthenticated behind promoteToAuthenticated return value - Fix transport message handler: use tempConn directly when ws is in temp state, preventing stale _connections entry from causing promoteToAuthenticated to fail - Close competing temp connections with same identifier on promotion - Expose __yonexusServer on globalThis for cross-plugin communication - Remove auto-failure on admin notification miss; pairing stays pending Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
165
plugin/index.ts
165
plugin/index.ts
@@ -30,30 +30,173 @@ export {
|
||||
type YonexusServerStore
|
||||
} from "./core/store.js";
|
||||
|
||||
import path from "node:path";
|
||||
import fs from "node:fs";
|
||||
import { validateYonexusServerConfig } from "./core/config.js";
|
||||
import { createYonexusServerStore } from "./core/store.js";
|
||||
import { createServerTransport } from "./core/transport.js";
|
||||
import { createYonexusServerRuntime } from "./core/runtime.js";
|
||||
import { createServerRuleRegistry } from "./core/rules.js";
|
||||
import { encodeRuleMessage } from "../../Yonexus.Protocol/src/index.js";
|
||||
import type { ServerPersistenceData } from "./core/persistence.js";
|
||||
|
||||
export interface YonexusServerPluginManifest {
|
||||
readonly name: "Yonexus.Server";
|
||||
readonly version: string;
|
||||
readonly description: string;
|
||||
}
|
||||
|
||||
export interface YonexusServerPluginRuntime {
|
||||
readonly hooks: readonly [];
|
||||
readonly commands: readonly [];
|
||||
readonly tools: readonly [];
|
||||
}
|
||||
|
||||
const manifest: YonexusServerPluginManifest = {
|
||||
name: "Yonexus.Server",
|
||||
version: "0.1.0",
|
||||
description: "Yonexus central hub plugin for cross-instance OpenClaw communication"
|
||||
};
|
||||
|
||||
export function createYonexusServerPlugin(): YonexusServerPluginRuntime {
|
||||
return {
|
||||
hooks: [],
|
||||
commands: [],
|
||||
tools: []
|
||||
let _serverStarted = false;
|
||||
|
||||
export function createYonexusServerPlugin(api: {
|
||||
rootDir: string;
|
||||
pluginConfig: unknown;
|
||||
registrationMode?: string; // "full" (gateway) | "cli-metadata" | "setup-only" | "setup-runtime"
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
registerCli?: (registrar: (ctx: { program: any }) => void, opts?: { commands?: string[] }) => void;
|
||||
}): void {
|
||||
const stateFilePath = path.join(api.rootDir, "state.json");
|
||||
|
||||
// Register CLI regardless of whether the gateway is already running.
|
||||
// The CLI process is a separate invocation that reads from the persisted state file.
|
||||
api.registerCli?.(({ program }) => {
|
||||
const group = program
|
||||
.command("yonexus-server")
|
||||
.description("Yonexus.Server management");
|
||||
|
||||
group
|
||||
.command("pair-code <identifier>")
|
||||
.description("Show the pending pairing code for a device awaiting confirmation")
|
||||
.action((identifier: string) => {
|
||||
let raw: ServerPersistenceData;
|
||||
try {
|
||||
raw = JSON.parse(fs.readFileSync(stateFilePath, "utf8")) as ServerPersistenceData;
|
||||
} catch {
|
||||
console.error("Error: could not read server state. Is the gateway running?");
|
||||
process.exit(1);
|
||||
}
|
||||
|
||||
const client = raw.clients?.find((c) => c.identifier === identifier);
|
||||
if (!client) {
|
||||
console.error(`Error: identifier "${identifier}" not found in server registry.`);
|
||||
process.exit(1);
|
||||
}
|
||||
if (client.pairingStatus !== "pending" || !client.pairingCode) {
|
||||
const status = client.pairingStatus;
|
||||
console.error(`Error: no pending pairing for "${identifier}" (status: ${status}).`);
|
||||
process.exit(1);
|
||||
}
|
||||
if (client.pairingExpiresAt && Math.floor(Date.now() / 1000) > client.pairingExpiresAt) {
|
||||
console.error(`Error: pairing for "${identifier}" has expired.`);
|
||||
process.exit(1);
|
||||
}
|
||||
|
||||
const expiresIn = client.pairingExpiresAt
|
||||
? Math.max(0, client.pairingExpiresAt - Math.floor(Date.now() / 1000))
|
||||
: 0;
|
||||
const mm = String(Math.floor(expiresIn / 60)).padStart(2, "0");
|
||||
const ss = String(expiresIn % 60).padStart(2, "0");
|
||||
|
||||
console.log(`Identifier : ${client.identifier}`);
|
||||
console.log(`Pairing code : ${client.pairingCode}`);
|
||||
console.log(`Expires in : ${mm}m ${ss}s`);
|
||||
});
|
||||
|
||||
group
|
||||
.command("list-pending")
|
||||
.description("List all identifiers with a pending pairing code")
|
||||
.action(() => {
|
||||
let raw: ServerPersistenceData;
|
||||
try {
|
||||
raw = JSON.parse(fs.readFileSync(stateFilePath, "utf8")) as ServerPersistenceData;
|
||||
} catch {
|
||||
console.error("Error: could not read server state. Is the gateway running?");
|
||||
process.exit(1);
|
||||
}
|
||||
|
||||
const now = Math.floor(Date.now() / 1000);
|
||||
const pending = (raw.clients ?? []).filter(
|
||||
(c) => c.pairingStatus === "pending" && c.pairingCode && (!c.pairingExpiresAt || now <= c.pairingExpiresAt)
|
||||
);
|
||||
|
||||
if (pending.length === 0) {
|
||||
console.log("No pending pairings.");
|
||||
return;
|
||||
}
|
||||
for (const c of pending) {
|
||||
const expiresIn = c.pairingExpiresAt ? Math.max(0, c.pairingExpiresAt - now) : 0;
|
||||
const mm = String(Math.floor(expiresIn / 60)).padStart(2, "0");
|
||||
const ss = String(expiresIn % 60).padStart(2, "0");
|
||||
console.log(` ${c.identifier} (expires in ${mm}m ${ss}s)`);
|
||||
}
|
||||
});
|
||||
}, { commands: ["yonexus-server"] });
|
||||
|
||||
if (_serverStarted) return;
|
||||
_serverStarted = true;
|
||||
|
||||
const config = validateYonexusServerConfig(api.pluginConfig);
|
||||
const store = createYonexusServerStore(stateFilePath);
|
||||
|
||||
const ruleRegistry = createServerRuleRegistry();
|
||||
const onClientAuthenticatedCallbacks: Array<(identifier: string) => void> = [];
|
||||
|
||||
let runtimeRef: ReturnType<typeof createYonexusServerRuntime> | null = null;
|
||||
const transport = createServerTransport({
|
||||
config,
|
||||
onMessage: (conn, msg) => {
|
||||
runtimeRef?.handleMessage(conn, msg).catch((err: unknown) => {
|
||||
console.error("[yonexus-server] message handler error:", err);
|
||||
});
|
||||
},
|
||||
onDisconnect: (identifier) => {
|
||||
if (identifier && runtimeRef) {
|
||||
runtimeRef.handleDisconnect(identifier);
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
// Expose registry and helpers for other plugins loaded in the same process
|
||||
(globalThis as Record<string, unknown>)["__yonexusServer"] = {
|
||||
ruleRegistry,
|
||||
sendRule: (identifier: string, ruleId: string, content: string): boolean =>
|
||||
transport.send(identifier, encodeRuleMessage(ruleId, content)),
|
||||
onClientAuthenticated: onClientAuthenticatedCallbacks
|
||||
};
|
||||
|
||||
const runtime = createYonexusServerRuntime({
|
||||
config,
|
||||
store,
|
||||
transport,
|
||||
ruleRegistry,
|
||||
onClientAuthenticated: (identifier) => {
|
||||
for (const cb of onClientAuthenticatedCallbacks) cb(identifier);
|
||||
}
|
||||
});
|
||||
runtimeRef = runtime;
|
||||
|
||||
const shutdown = (): void => {
|
||||
runtime.stop().catch((err: unknown) => {
|
||||
console.error("[yonexus-server] shutdown error:", err);
|
||||
});
|
||||
};
|
||||
process.once("SIGTERM", shutdown);
|
||||
process.once("SIGINT", shutdown);
|
||||
|
||||
runtime.start().catch((err: unknown) => {
|
||||
// EADDRINUSE means the gateway is already running (e.g. this is a CLI invocation).
|
||||
// Any other error is a real problem worth logging.
|
||||
const code = (err as NodeJS.ErrnoException | undefined)?.code;
|
||||
if (code !== "EADDRINUSE") {
|
||||
console.error("[yonexus-server] failed to start:", err);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
export default createYonexusServerPlugin;
|
||||
|
||||
Reference in New Issue
Block a user