fix: remove rxjs (#376)
Some checks are pending
Continuous Delivery / Publishing Dev (push) Waiting to run
NPM / Publish / test-and-publish (push) Waiting to run

* firstcommit

* removerxjs

* document-task

* documentation+clean

* fixregres

* fix+regress

* fix+regres+errorhandling
This commit is contained in:
Jacob Nguyen
2025-01-13 10:33:53 -06:00
committed by GitHub
parent 7deb79e907
commit 59d08ef207
18 changed files with 970 additions and 906 deletions

View File

@@ -1,219 +1,18 @@
import type { Interaction, Message, BaseInteraction } from 'discord.js';
import util from 'node:util';
import {
EMPTY, type Observable, concatMap, filter,
throwError, fromEvent, map, type OperatorFunction,
catchError, finalize, pipe, from, take, share, of,
} from 'rxjs';
import * as Id from '../core/id'
import type { Emitter, ErrorHandling, Logging } from '../core/interfaces';
import type { Emitter, Logging } from '../core/interfaces';
import { SernError } from '../core/structures/enums'
import { EMPTY_ERR, Err, Ok, Result, wrapAsync } from '../core/structures/result';
import type { UnpackedDependencies } from '../types/utility';
import type { CommandModule, Module, Processed } from '../types/core-modules';
import * as assert from 'node:assert';
import { Context } from '../core/structures/context';
import { CommandType } from '../core/structures/enums'
import { Ok, wrapAsync} from '../core/structures/result';
import type { Module } from '../types/core-modules';
import { inspect } from 'node:util'
import { disposeAll } from '../core/ioc';
import { resultPayload, isAutocomplete, treeSearch, fmt } from '../core/functions'
import { resultPayload } from '../core/functions'
import merge from 'deepmerge'
function handleError<C>(crashHandler: ErrorHandling, emitter: Emitter, logging?: Logging) {
return (pload: unknown, caught: Observable<C>) => {
// This is done to fit the ErrorHandling contract
if(!emitter.emit('error', pload)) {
const err = pload instanceof Error ? pload : Error(util.inspect(pload, { colors: true }));
logging?.error({ message: util.inspect(pload) });
crashHandler.updateAlive(err);
}
return caught;
};
}
const arrayify= <T>(src: T) =>
Array.isArray(src) ? src : [src];
interface ExecutePayload {
module: Module;
args: unknown[];
deps: Dependencies;
params?: string;
[key: string]: unknown
}
export const filterTap = <K, R>(onErr: (e: R) => void): OperatorFunction<Result<K, R>, K> =>
concatMap(result => {
if(result.ok){
return of(result.value)
}
onErr(result.error);
return EMPTY;
})
export const sharedEventStream = <T>(e: Emitter, eventName: string) =>
(fromEvent(e, eventName) as Observable<T>).pipe(share());
function intoPayload(module: Module, deps: Dependencies) {
return pipe(map(arrayify),
map(args => ({ module, args, deps })),
map(p => p.args));
}
/**
* Creates an observable from { source }
* @param module
* @param source
*/
export function eventDispatcher(deps: Dependencies, module: Module, source: unknown) {
assert.ok(source && typeof source === 'object',
`${source} cannot be constructed into an event listener`);
const execute: OperatorFunction<unknown[]|undefined, unknown> =
concatMap(async args => {
if(args) return Reflect.apply(module.execute, null, args);
});
//@ts-ignore
let ev = fromEvent(source ,module.name!);
//@ts-ignore
if(module['once']) {
ev = ev.pipe(take(1))
}
return ev.pipe(intoPayload(module, deps),
execute);
}
interface DispatchPayload {
module: Processed<CommandModule>;
event: BaseInteraction;
defaultPrefix?: string;
deps: Dependencies;
params?: string
};
export function createDispatcher({ module, event, defaultPrefix, deps, params }: DispatchPayload): ExecutePayload {
assert.ok(CommandType.Text !== module.type,
SernError.MismatchEvent + 'Found text command in interaction stream');
if(isAutocomplete(event)) {
assert.ok(module.type === CommandType.Slash
|| module.type === CommandType.Both, "Autocomplete option on non command interaction");
const option = treeSearch(event, module.options);
assert.ok(option, SernError.NotSupportedInteraction + ` There is no autocomplete tag for ` + inspect(module));
const { command } = option;
return { module: command as Processed<Module>, //autocomplete is not a true "module" warning cast!
args: [event],
deps };
}
switch (module.type) {
case CommandType.Slash:
case CommandType.Both: {
return { module, args: [Context.wrap(event, defaultPrefix)], deps };
}
default: return { module, args: [event], deps, params };
}
}
function createGenericHandler<Source, Narrowed extends Source, Output>(
source: Observable<Source>,
makeModule: (event: Narrowed) => Promise<Output>,
) {
return (pred: (i: Source) => i is Narrowed) =>
source.pipe(
filter(pred), // only handle this stream if it passes pred
concatMap(makeModule)); // create a payload, preparing to execute
}
/**
*
* Creates an RxJS observable that filters and maps incoming interactions to their respective modules.
* @param i An RxJS observable of interactions.
* @param mg The module manager instance used to retrieve the module path for each interaction.
* @returns A handler to create a RxJS observable of dispatchers that take incoming interactions and execute their corresponding modules.
*/
export function createInteractionHandler<T extends Interaction>(
source: Observable<Interaction>,
deps: Dependencies,
defaultPrefix?: string
) {
const mg = deps['@sern/modules'];
return createGenericHandler<Interaction, T, Result<ReturnType<typeof createDispatcher>, void>>(
source,
async event => {
const possibleIds = Id.reconstruct(event);
let modules = possibleIds
.map(({ id, params }) => ({ module: mg.get(id), params }))
.filter(({ module }) => module !== undefined);
if(modules.length == 0) {
return EMPTY_ERR;
}
const [{module, params}] = modules;
return Ok(createDispatcher({
module: module as Processed<CommandModule>,
event, defaultPrefix, deps, params
}));
});
}
export function createMessageHandler(
source: Observable<Message>,
defaultPrefix: string,
deps: Dependencies,
) {
const mg = deps['@sern/modules'];
return createGenericHandler(source, async event => {
const [prefix] = fmt(event.content, defaultPrefix);
let module= mg.get(`${prefix}_T`) ?? mg.get(`${prefix}_B`) as Module;
if(!module) {
return Err('Possibly undefined behavior: could not find a static id to resolve');
}
return Ok({ args: [Context.wrap(event, defaultPrefix)], module, deps })
});
}
/**
* Wraps the task in a Result as a try / catch.
* if the task is ok, an event is emitted and the stream becomes empty
* if the task is an error, throw an error down the stream which will be handled by catchError
* thank u kingomes
* @param emitter reference to SernEmitter that will emit a successful execution of module
* @param module the module that will be executed with task
* @param task the deferred execution which will be called
*/
export function executeModule(emitter: Emitter, { module, args }: ExecutePayload) {
return from(wrapAsync(async () => module.execute(...args)))
.pipe(concatMap(result => {
if (result.ok){
emitter.emit('module.activate', resultPayload('success', module));
return EMPTY;
}
return throwError(() => resultPayload('failure', module, result.error));
}))
};
/**
* A higher order function that
* - calls all control plugins.
* - any failures results to { config.onStop } being called
* - if all results are ok, the stream is converted to { config.onNext }
* config.onNext will be returned if everything is okay.
* @param config
* @returns function which calls all plugins and returns onNext or fail
*/
export function createResultResolver<Output>(config: {
onStop?: (module: Module, err?: string) => unknown;
onNext: (args: ExecutePayload, map: Record<string, unknown>) => Output;
}) {
const { onStop, onNext } = config;
return async (payload: ExecutePayload) => {
const task = await callPlugins(payload);
if (!task) throw Error("Plugin did not return anything.");
if(!task.ok) {
onStop?.(payload.module, String(task.error));
} else {
return onNext(payload, task.value) as Output;
}
};
};
function isObject(item: unknown) {
return (item && typeof item === 'object' && !Array.isArray(item));
@@ -238,20 +37,32 @@ export async function callInitPlugins(_module: Module, deps: Dependencies, emit?
return module
}
export async function callPlugins({ args, module, deps, params }: ExecutePayload) {
export function executeModule(emitter: Emitter, logger: Logging|undefined, { module, args } : ExecutePayload) {
const moduleCalled = wrapAsync(async () => {
return module.execute(...args);
})
moduleCalled
.then((res) => {
if(res.ok) {
emitter.emit('module.activate', resultPayload('success', module))
} else {
if(!emitter.emit('error', resultPayload('failure', module, res.error))) {
// node crashes here.
logger?.error({ 'message': res.error })
}
}
})
.catch(err => {
throw err
})
};
export async function callPlugins({ args, module }: ExecutePayload) {
let state = {};
for(const plugin of module.onEvent??[]) {
const executionContext = {
state,
deps,
params,
type: module.type,
module: { name: module.name,
description: module.description,
locals: module.locals,
meta: module.meta }
};
const result = await plugin.execute(...args, executionContext);
const result = await plugin.execute(...args);
if(!result.ok) {
return result;
}
@@ -261,34 +72,3 @@ export async function callPlugins({ args, module, deps, params }: ExecutePayload
}
return Ok(state);
}
/**
* Creates an executable task ( execute the command ) if all control plugins are successful
* this needs to go
* @param onStop emits a failure response to the SernEmitter
*/
export function intoTask(onStop: (m: Module) => unknown) {
const onNext = ({ args, module, deps, params }: ExecutePayload, state: Record<string, unknown>) => {
return {
module,
args: [...args, { state,
deps,
params,
type: module.type,
module: { name: module.name,
description: module.description,
locals: module.locals,
meta: module.meta } }],
deps
}
};
return createResultResolver({ onStop, onNext });
}
export const handleCrash = ({ "@sern/errors": err, '@sern/emitter': sem, '@sern/logger': log } : UnpackedDependencies, metadata: string) =>
pipe(catchError(handleError(err, sem, log)),
finalize(() => {
log?.info({ message: 'A stream closed: ' + metadata });
disposeAll(log);
}))

View File

@@ -1,30 +1,58 @@
import type { Interaction } from 'discord.js';
import { mergeMap, merge, concatMap, EMPTY } from 'rxjs';
import { createInteractionHandler, executeModule, intoTask, sharedEventStream, filterTap, handleCrash } from './event-utils';
import type { Module } from '../types/core-modules'
import { callPlugins, executeModule } from './event-utils';
import { SernError } from '../core/structures/enums'
import { isAutocomplete, isCommand, isMessageComponent, isModal, resultPayload } from '../core/functions'
import { UnpackedDependencies } from '../types/utility';
import { createSDT, isAutocomplete, isCommand, isMessageComponent, isModal, resultPayload, treeSearch } from '../core/functions'
import type { UnpackedDependencies } from '../types/utility';
import * as Id from '../core/id'
import { Context } from '../core/structures/context';
export default function interactionHandler(deps: UnpackedDependencies, defaultPrefix?: string) {
export function interactionHandler(deps: UnpackedDependencies, defaultPrefix?: string) {
//i wish javascript had clojure destructuring
const { '@sern/client': client,
'@sern/emitter': emitter } = deps
const interactionStream$ = sharedEventStream<Interaction>(client, 'interactionCreate');
const handle = createInteractionHandler(interactionStream$, deps, defaultPrefix);
'@sern/modules': moduleManager,
'@sern/logger': log,
'@sern/emitter': reporter } = deps
const interactionHandler$ = merge(handle(isMessageComponent),
handle(isAutocomplete),
handle(isCommand),
handle(isModal));
return interactionHandler$
.pipe(filterTap(e => emitter.emit('warning', resultPayload('warning', undefined, e))),
concatMap(intoTask(module => {
emitter.emit('module.activate', resultPayload('failure', module, SernError.PluginFailure))
})),
mergeMap(payload => {
if(payload)
return executeModule(emitter, payload)
return EMPTY;
}),
handleCrash(deps, "interaction handling"));
client.on('interactionCreate', async (event) => {
//returns array of possible ids
const possibleIds = Id.reconstruct(event);
let modules = possibleIds
.map(({ id, params }) => ({ module: moduleManager.get(id)!, params }))
.filter(({ module }) => module !== undefined);
if(modules.length == 0) {
return;
}
const { module, params } = modules.at(0)!;
let payload;
if(isAutocomplete(event)) {
//@ts-ignore stfu
const { command } = treeSearch(event, module.options);
payload= { module: command as Module, //autocomplete is not a true "module" warning cast!
args: [event, createSDT(command, deps, params)] };
} else if(isCommand(event)) {
payload= { module, args: [Context.wrap(event, defaultPrefix), createSDT(module, deps, params)] };
} else if (isModal(event) || isMessageComponent(event)) {
payload= { module, args: [event, createSDT(module, deps, params)] }
} else {
throw Error("Unknown interaction while handling in interactionCreate event " + event)
}
const result = await callPlugins(payload)
if(!result.ok) {
reporter.emit('module.activate', resultPayload('failure', module, result.error ?? SernError.PluginFailure))
return
}
if(payload.args.length !== 2) {
throw Error ('Invalid payload')
}
//@ts-ignore assigning final state from plugin
payload.args[1].state = result.value
// note: do not await this. will be blocking if long task (ie waiting for modal input)
executeModule(reporter, log, payload);
});
}

View File

@@ -1,17 +1,17 @@
import { EMPTY, mergeMap, concatMap } from 'rxjs';
import type { Message } from 'discord.js';
import { createMessageHandler, executeModule, intoTask, sharedEventStream, filterTap, handleCrash} from './event-utils';
import { callPlugins, executeModule } from './event-utils';
import { SernError } from '../core/structures/enums'
import { resultPayload } from '../core/functions'
import { UnpackedDependencies } from '../types/utility';
import type { Emitter } from '../core/interfaces';
import { createSDT, fmt, resultPayload } from '../core/functions'
import type { UnpackedDependencies } from '../types/utility';
import type { Module } from '../types/core-modules';
import { Context } from '../core/structures/context';
/**
* Ignores messages from any person / bot except itself
* @param prefix
*/
function isNonBot(prefix: string) {
return (msg: Message): msg is Message => !msg.author.bot && hasPrefix(prefix, msg.content);
function isBotOrNoPrefix(msg: Message, prefix: string) {
return msg.author.bot || !hasPrefix(prefix, msg.content);
}
function hasPrefix(prefix: string, content: string) {
@@ -19,32 +19,36 @@ function hasPrefix(prefix: string, content: string) {
return prefixInContent.localeCompare(prefix, undefined, { sensitivity: 'accent' }) === 0;
}
export default
function (deps: UnpackedDependencies, defaultPrefix?: string) {
export function messageHandler (deps: UnpackedDependencies, defaultPrefix?: string) {
const {"@sern/emitter": emitter,
'@sern/logger': log,
'@sern/modules': mg,
'@sern/client': client} = deps
if (!defaultPrefix) {
log?.debug({ message: 'No prefix found. message handler shutting down' });
return EMPTY;
return;
}
const messageStream$ = sharedEventStream<Message>(client as unknown as Emitter, 'messageCreate');
const handle = createMessageHandler(messageStream$, defaultPrefix, deps);
client.on('messageCreate', async message => {
if(isBotOrNoPrefix(message, defaultPrefix)) {
return
}
const [prefix] = fmt(message.content, defaultPrefix);
let module = mg.get(`${prefix}_T`) ?? mg.get(`${prefix}_B`) as Module;
if(!module) {
throw Error('Possibly undefined behavior: could not find a static id to resolve')
}
const payload = { module, args: [Context.wrap(message, defaultPrefix), createSDT(module, deps, undefined)] }
const result = await callPlugins(payload)
if (!result.ok) {
emitter.emit('module.activate', resultPayload('failure', module, result.error ?? SernError.PluginFailure))
return
}
const msgCommands$ = handle(isNonBot(defaultPrefix));
//@ts-ignore
payload.args[1].state = result.value
return msgCommands$.pipe(
filterTap(e => emitter.emit('warning', resultPayload('warning', undefined, e))),
concatMap(intoTask(module => {
const result = resultPayload('failure', module, SernError.PluginFailure);
emitter.emit('module.activate', result);
})),
mergeMap(payload => {
if(payload)
return executeModule(emitter, payload)
return EMPTY;
}),
handleCrash(deps, "message handling")
)
executeModule(emitter, log, payload)
})
}

View File

@@ -1,40 +1,99 @@
import { concatMap, from, interval, of, map, startWith, fromEvent, take, mergeScan } from "rxjs"
import { Presence } from "../core/presences";
import { Services } from "../core/ioc";
import assert from "node:assert";
import * as Files from "../core/module-loading";
type SetPresence = (conf: Presence.Result) => Promise<unknown>
const parseConfig = async (conf: Promise<Presence.Result>) => {
return conf.then(s => {
if('repeat' in s) {
const { onRepeat, repeat } = s;
assert(repeat !== undefined, "repeat option is undefined");
assert(onRepeat !== undefined, "onRepeat callback is undefined, but repeat exists");
const src$ = typeof repeat === 'number'
? interval(repeat)
: fromEvent(...repeat);
return src$.pipe(mergeScan(async (args) => onRepeat(args), s),
startWith(s));
const parseConfig = async (conf: Promise<Presence.Result>, setPresence: SetPresence) => {
const result = await conf;
if ('repeat' in result) {
const { onRepeat, repeat } = result;
// Validate configuration
if (repeat === undefined) {
throw new Error("repeat option is undefined");
}
return of(s).pipe(take(1));
})
if (onRepeat === undefined) {
throw new Error("onRepeat callback is undefined, but repeat exists");
}
// Initial state
let currentState = result;
const processState = async (state: typeof currentState) => {
try {
const result = onRepeat(state);
// If it's a promise, await it, otherwise use the value directly
return result instanceof Promise ? await result : result;
} catch (error) {
// TODO process error
//console.error(error);
return state; // Return previous state on error
}
};
// Handle numeric interval
if (typeof repeat === 'number') {
// Return a promise that never resolves (or resolves on cleanup)
return new Promise((resolve) => {
// Immediately return initial state
processState(currentState);
// Set up interval
let isProcessing = false;
const intervalId = setInterval(() => {
// Skip if previous operation is still running
if (isProcessing) return;
isProcessing = true;
processState(currentState)
.then(newState => {
currentState = newState;
return setPresence(currentState)
})
.catch(console.error)
.finally(() => {
isProcessing = false;
});
}, repeat);
// Optional: Return cleanup function
return () => clearInterval(intervalId);
});
}
// Handle event-based repeat
else {
const handler = async () => {
currentState = await onRepeat(currentState);
await setPresence(currentState);
};
let has_registered = false;
return new Promise((resolve) => {
const [target, eventName] = repeat;
// Immediately return initial state
processState(currentState);
// Set up event listener
if(!has_registered) {
target.addListener(eventName, handler);
has_registered=true;
}
// Optional: Return cleanup function
return () => target.removeListener(eventName, handler);
});
}
}
// No repeat configuration, just return the result
return setPresence(result);
};
export const presenceHandler = (path: string, setPresence: SetPresence) => {
const presence = Files
.importModule<Presence.Config<(keyof Dependencies)[]>>(path)
.then(({ module }) => {
//fetch services with the order preserved, passing it to the execute fn
const fetchedServices = Services(...module.inject ?? []);
return async () => module.execute(...fetchedServices);
})
const module$ = from(presence);
return module$.pipe(
//compose:
//call the execute function, passing that result into parseConfig.
//concatMap resolves the promise, and passes it to the next concatMap.
concatMap(fn => parseConfig(fn())),
// subscribe to the observable parseConfig yields, and set the presence.
concatMap(conf => conf.pipe(map(setPresence))));
export const presenceHandler = async (path: string, setPresence: SetPresence) => {
const presence = await
Files.importModule<Presence.Config<(keyof Dependencies)[]>>(path)
.then(({ module }) => {
//fetch services with the order preserved, passing it to the execute fn
const fetchedServices = Services(...module.inject ?? []);
return async () => module.execute(...fetchedServices);
})
return parseConfig(presence(), setPresence);
}

View File

@@ -3,7 +3,7 @@ import { once } from 'node:events';
import { resultPayload } from '../core/functions';
import { CommandType } from '../core/structures/enums';
import { Module } from '../types/core-modules';
import { UnpackedDependencies } from '../types/utility';
import type { UnpackedDependencies } from '../types/utility';
import { callInitPlugins } from './event-utils';
export default async function(dir: string, deps : UnpackedDependencies) {

View File

@@ -1,33 +1,57 @@
import { EventType, SernError } from '../core/structures/enums';
import { callInitPlugins, eventDispatcher, handleCrash } from './event-utils'
import { callInitPlugins } from './event-utils'
import { EventModule, Module } from '../types/core-modules';
import * as Files from '../core/module-loading'
import type { UnpackedDependencies } from '../types/utility';
import { from, map, mergeAll } from 'rxjs';
import type { Emitter } from '../core/interfaces';
import { inspect } from 'util'
import { resultPayload } from '../core/functions';
import type { Wrapper } from '../'
const intoDispatcher = (deps: UnpackedDependencies) =>
(module : EventModule) => {
switch (module.type) {
case EventType.Sern:
return eventDispatcher(deps, module, deps['@sern/emitter']);
case EventType.Discord:
return eventDispatcher(deps, module, deps['@sern/client']);
case EventType.External:
return eventDispatcher(deps, module, deps[module.emitter]);
default: throw Error(SernError.InvalidModuleType + ' while creating event handler');
}
};
export default async function(deps: UnpackedDependencies, eventDir: string) {
export default async function(deps: UnpackedDependencies, wrapper: Wrapper) {
const eventModules: EventModule[] = [];
for await (const path of Files.readRecursive(eventDir)) {
for await (const path of Files.readRecursive(wrapper.events!)) {
let { module } = await Files.importModule<Module>(path);
await callInitPlugins(module, deps)
eventModules.push(module as EventModule);
}
from(eventModules)
.pipe(map(intoDispatcher(deps)),
mergeAll(), // all eventListeners are turned on
handleCrash(deps, "event modules"))
.subscribe();
const logger = deps['@sern/logger'], report = deps['@sern/emitter'];
for (const module of eventModules) {
let source: Emitter;
switch (module.type) {
case EventType.Sern:
source=deps['@sern/emitter'];
break
case EventType.Discord:
source=deps['@sern/client'];
break
case EventType.External:
source=deps[module.emitter] as Emitter;
break
default: throw Error(SernError.InvalidModuleType + ' while creating event handler');
}
if(!source && typeof source !== 'object') {
throw Error(`${source} cannot be constructed into an event listener`)
}
if(!('addListener' in source && 'removeListener' in source)) {
throw Error('source must implement Emitter')
}
const execute = async (...args: any[]) => {
try {
if(args) {
if('once' in module) { source.removeListener(String(module.name!), execute); }
await Reflect.apply(module.execute, null, args);
}
} catch(e) {
const err = e instanceof Error ? e : Error(inspect(e, { colors: true }));
if(!report.emit('error', resultPayload('failure', module, err))) {
logger?.error({ message: inspect(err) });
}
}
}
source.addListener(String(module.name!), execute)
}
}