File size: 9,355 Bytes
bc4a7e8 | 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 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 | /**
* Custom hook registry β event-driven plugin hook system.
*
* Plugins can register handlers for any OmniRoute event. Built-in events
* cover the full request lifecycle plus routing, rate limiting, and errors.
*
* @module plugins/hooks
*/
import { logger } from "../../../open-sse/utils/logger.ts";
const log = logger("PLUGIN_HOOKS");
// ββ Types ββ
export type BlockingHookResult = {
blocked?: boolean;
response?: unknown;
body?: unknown;
metadata?: Record<string, unknown>;
};
export type HookHandler = (
payload: unknown
) => void | Promise<void> | BlockingHookResult | Promise<BlockingHookResult>;
export interface HookRegistration {
pluginName: string;
handler: HookHandler;
priority: number;
}
// ββ Built-in events ββ
export const BUILTIN_EVENTS = [
"onRequest",
"onResponse",
"onError",
"onModelSelect",
"onComboResolve",
"onRateLimit",
"onQuotaExhaust",
"onProviderError",
"onStreamStart",
"onStreamEnd",
] as const;
export type BuiltinEvent = (typeof BUILTIN_EVENTS)[number];
// ββ Rate limiting ββ
const RATE_LIMIT_MAX = 100; // max calls per plugin per window
const RATE_LIMIT_WINDOW_MS = 1000; // 1 second window
interface RateLimitState {
count: number;
windowStart: number;
}
const rateLimitMap: Map<string, RateLimitState> = new Map();
function isRateLimited(pluginName: string): boolean {
const now = Date.now();
const key = pluginName;
const state = rateLimitMap.get(key);
if (!state || now - state.windowStart >= RATE_LIMIT_WINDOW_MS) {
// New window
rateLimitMap.set(key, { count: 1, windowStart: now });
return false;
}
state.count++;
if (state.count > RATE_LIMIT_MAX) {
return true;
}
return false;
}
// ββ Registry ββ
const hooks: Map<string, HookRegistration[]> = new Map();
/**
* Register a handler for an event.
*/
export function registerHook(
event: string,
pluginName: string,
handler: HookHandler,
priority: number = 100
): void {
if (!hooks.has(event)) {
hooks.set(event, []);
}
const list = hooks.get(event)!;
// Prevent duplicate registration
if (list.some((r) => r.pluginName === pluginName && r.handler === handler)) {
return;
}
list.push({ pluginName, handler, priority });
list.sort((a, b) => a.priority - b.priority);
log.info("hook.registered", { event, pluginName, priority });
}
/**
* Unregister all handlers for a plugin.
* Also evicts the plugin's rate-limit state so uninstalled plugins don't leak memory.
*/
export function unregisterHooks(pluginName: string): void {
for (const [event, list] of hooks.entries()) {
const before = list.length;
const filtered = list.filter((r) => r.pluginName !== pluginName);
if (filtered.length !== before) {
hooks.set(event, filtered);
log.info("hook.unregistered", { event, pluginName, removed: before - filtered.length });
}
}
// Evict rate-limit state so uninstalled plugins don't accumulate entries
rateLimitMap.delete(pluginName);
}
/**
* Unregister a specific handler.
*/
export function unregisterHook(event: string, pluginName: string): void {
const list = hooks.get(event);
if (!list) return;
const before = list.length;
const filtered = list.filter((r) => r.pluginName !== pluginName);
hooks.set(event, filtered);
if (before !== filtered.length) {
log.info("hook.unregistered", { event, pluginName });
}
}
/**
* Emit an event β fire all registered handlers.
* Handler errors are logged but don't block other handlers.
* Rate-limited per plugin: max 100 calls per second.
*/
export async function emitHook(event: string, payload: unknown): Promise<void> {
const list = hooks.get(event);
if (!list || list.length === 0) return;
for (const reg of list) {
if (isRateLimited(reg.pluginName)) {
log.warn("hook.rate_limited", { event, pluginName: reg.pluginName });
continue;
}
try {
await reg.handler(payload);
} catch (err: unknown) {
const message = err instanceof Error ? err.message : String(err);
log.error("hook.handler_error", {
event,
pluginName: reg.pluginName,
error: message,
});
}
}
}
/**
* Emit a blocking event β fire handlers with body/metadata chaining.
* Returns blocking result from the first handler that blocks, or merged body/metadata.
* Used for onRequest and onResponse where plugins can modify or block the request.
*/
export async function emitHookBlocking(
event: string,
payload: unknown
): Promise<{
blocked?: boolean;
response?: unknown;
body?: unknown;
metadata?: Record<string, unknown>;
}> {
const list = hooks.get(event) || [];
const ctx = (payload || {}) as Record<string, unknown>;
let mergedBody: unknown = ctx.body;
let mergedMetadata: Record<string, unknown> = (ctx.metadata as Record<string, unknown>) || {};
for (const reg of list) {
// Mirror emitHook: rate-limit the hot blocking path too
if (isRateLimited(reg.pluginName)) {
log.warn("hook.blocking_rate_limited", { event, pluginName: reg.pluginName });
continue;
}
try {
// Chain the payload: each handler must see the body/metadata as mutated by
// previous handlers, not the original static payload β otherwise plugin B
// can't observe plugin A's changes. (#3286)
const currentPayload = { ...ctx, body: mergedBody, metadata: mergedMetadata };
const result = await reg.handler(currentPayload);
if (result && typeof result === "object") {
if ("body" in result) mergedBody = (result as Record<string, unknown>).body;
if ("metadata" in result)
mergedMetadata = {
...mergedMetadata,
...(((result as Record<string, unknown>).metadata as Record<string, unknown>) || {}),
};
if ("blocked" in result && (result as BlockingHookResult).blocked) {
return {
...result,
body: (result as BlockingHookResult).body ?? mergedBody,
metadata: { ...mergedMetadata, ...((result as BlockingHookResult).metadata || {}) },
};
}
}
} catch (err: unknown) {
const message = err instanceof Error ? err.message : String(err);
log.error("hook.blocking_handler_error", {
event,
pluginName: reg.pluginName,
error: message,
});
}
}
return { body: mergedBody, metadata: mergedMetadata };
}
// ββ Lifecycle wrappers (for chatCore.ts convenience) ββ
export interface PluginContext {
requestId: string;
body: unknown;
model: string;
provider: string;
apiKeyInfo?: unknown;
metadata: Record<string, unknown>;
}
export interface PluginResult {
blocked?: boolean;
response?: unknown;
body?: unknown;
metadata?: Record<string, unknown>;
}
// ββ Plugin interface (for loader/manager compatibility) ββ
export interface Plugin {
name: string;
priority?: number;
enabled?: boolean;
onRequest?: (ctx: PluginContext) => Promise<PluginResult | void> | PluginResult | void;
onResponse?: (ctx: PluginContext, response: unknown) => Promise<unknown | void> | unknown | void;
onError?: (ctx: PluginContext, error: Error) => Promise<unknown | void> | unknown | void;
}
/**
* Run onRequest hooks β blocking. Plugins can modify body/metadata or block with 403.
*/
export async function runOnRequest(ctx: PluginContext): Promise<PluginResult> {
return emitHookBlocking("onRequest", ctx);
}
/**
* Run onResponse hooks β chains response through plugins. Each plugin can modify the response.
*/
export async function runOnResponse(ctx: PluginContext, response: unknown): Promise<unknown> {
let currentResponse = response;
const list = hooks.get("onResponse") || [];
for (const reg of list) {
try {
const result = await reg.handler({ ...ctx, response: currentResponse });
if (
result !== undefined &&
result !== null &&
typeof result === "object" &&
"response" in result
) {
currentResponse = (result as { response: unknown }).response;
}
} catch (err: unknown) {
const message = err instanceof Error ? err.message : String(err);
log.error("hook.response_handler_error", { pluginName: reg.pluginName, error: message });
}
}
return currentResponse;
}
/**
* Run onError hooks β fire-and-forget notification.
*/
export async function runOnError(ctx: PluginContext, error: Error): Promise<void> {
await emitHook("onError", { ...ctx, error });
}
/**
* Get all registered hooks for an event.
*/
export function getHooks(event: string): HookRegistration[] {
return hooks.get(event) ?? [];
}
/**
* Get all events that have registered handlers.
*/
export function getActiveEvents(): string[] {
return [...hooks.entries()].filter(([, list]) => list.length > 0).map(([event]) => event);
}
/**
* Reset all hooks and rate limit state (for testing).
*/
export function resetHooks(): void {
hooks.clear();
rateLimitMap.clear();
}
|