fix+regres+errorhandling

package.json

@@ -38,8 +38,7 @@
         "@sern/ioc": "^1.1.0",
         "callsites": "^3.1.0",
         "cron": "^3.1.7",
-        "deepmerge": "^4.3.1",
+        "deepmerge": "^4.3.1"
-        "rxjs": "^7.8.0"
     },
     "devDependencies": {
         "@faker-js/faker": "^8.0.1",

src/core/functions.ts

@@ -11,7 +11,22 @@ import type {
 import { ApplicationCommandOptionType, InteractionType } from 'discord.js';
 import { PluginType } from './structures/enums';
 import assert from 'assert';
-import type { Payload } from '../types/utility';
+import type { Payload, UnpackedDependencies } from '../types/utility';
+
+export const createSDT = (module: Module, deps: UnpackedDependencies, params: string|undefined) => {
+    return {
+        state: {},
+        deps,
+        params, 
+        type: module.type,
+        module: {
+            name: module.name,
+            description: module.description,
+            locals: module.locals,
+            meta: module.meta
+        }
+    }
+}
 
 /**
  * Removes the first character(s) _[depending on prefix length]_ of the message

src/core/ioc.ts

@@ -70,7 +70,7 @@ export async function makeDependencies (conf: ValidDependencyConfig) {
     }
     container.addSingleton('@sern/errors', new __Services.DefaultErrorHandling);
     container.addSingleton('@sern/modules', new Map);
-    container.addSingleton('@sern/emitter', new EventEmitter)
+    container.addSingleton('@sern/emitter', new EventEmitter({ captureRejections: true }))
     container.addSingleton('@sern/scheduler', new __Services.TaskScheduler)
     conf(dependencyBuilder(container));
     await container.ready();

src/core/modules.ts

@@ -102,5 +102,34 @@ export function discordEvent<T extends keyof ClientEvents>(mod: {
     return eventModule({ type: EventType.Discord, ...mod, });
 }
 
-export function scheduledTask(ism: ScheduledTask) { return ism }
+/**
+ * Creates a scheduled task that can be executed at specified intervals using cron patterns
+ * 
+ * @param {ScheduledTask} ism - The scheduled task configuration object
+ * @param {string} ism.trigger - A cron pattern that determines when the task should execute
+ *                              Format: "* * * * *" (minute hour day month day-of-week)
+ * @param {Function} ism.execute - The function to execute when the task is triggered
+ * @param {Object} ism.execute.context - The execution context passed to the task
+ * 
+ * @returns {ScheduledTask} The configured scheduled task
+ * 
+ * @example
+ * // Create a task that runs every minute
+ * export default scheduledTask({
+ *   trigger: "* * * * *",
+ *   execute: (context) => {
+ *     console.log("Task executed!");
+ *   }
+ * });
+ * 
+ * @remarks
+ * - Tasks must be placed in the 'tasks' directory specified in your config
+ * - The file name serves as a unique identifier for the task
+ * - Tasks can be cancelled using deps['@sern/scheduler'].kill(uuid)
+ * 
+ * @see {@link https://crontab.guru/} for testing and creating cron patterns
+ */
+export function scheduledTask(ism: ScheduledTask): ScheduledTask { 
+    return ism 
+}
 

src/core/plugin.ts

@@ -15,16 +15,107 @@ export function makePlugin<V extends unknown[]>(
 export function EventInitPlugin(execute: (args: InitArgs) => PluginResult) {
     return makePlugin(PluginType.Init, execute);
 }
+
 /**
+ * Creates an initialization plugin for command preprocessing and modification
+ * 
  * @since 2.5.0
+ * @template I - Extends CommandType to enforce type safety for command modules
+ * 
+ * @param {function} execute - Function to execute during command initialization
+ * @param {InitArgs<T>} execute.args - The initialization arguments
+ * @param {T} execute.args.module - The command module being initialized
+ * @param {string} execute.args.absPath - The absolute path to the module file
+ * @param {Dependencies} execute.args.deps - Dependency injection container
+ * 
+ * @returns {Plugin} A plugin that runs during command initialization
+ * 
+ * @example
+ * // Plugin to update command description
+ * export const updateDescription = (description: string) => {
+ *   return CommandInitPlugin(({ deps }) => {
+ *     if(description.length > 100) {
+ *       deps.logger?.info({ message: "Invalid description" })
+ *       return controller.stop("From updateDescription: description is invalid");
+ *     }
+ *     module.description = description;
+ *     return controller.next();
+ *   });
+ * };
+ * 
+ * @example
+ * // Plugin to store registration date in module locals
+ * export const dateRegistered = () => {
+ *   return CommandInitPlugin(({ module }) => {
+ *     module.locals.registered = Date.now()
+ *     return controller.next();
+ *   });
+ * };
+ * 
+ * @remarks
+ * - Init plugins can modify how commands are loaded and perform preprocessing
+ * - The module.locals object can be used to store custom plugin-specific data
+ * - Be careful when modifying module fields as multiple plugins may interact with them
+ * - Use controller.next() to continue to the next plugin
+ * - Use controller.stop(reason) to halt plugin execution
  */
 export function CommandInitPlugin<I extends CommandType>(
     execute: (args: InitArgs) => PluginResult
-) {
+): Plugin {
     return makePlugin(PluginType.Init, execute);
 }
+
 /**
+ * Creates a control plugin for command preprocessing, filtering, and state management
+ * 
  * @since 2.5.0
+ * @template I - Extends CommandType to enforce type safety for command modules
+ * 
+ * @param {function} execute - Function to execute during command control flow
+ * @param {CommandArgs<I>} execute.args - The command arguments array
+ * @param {Context} execute.args[0] - The discord context (e.g., guild, channel, user info, interaction)
+ * @param {SDT} execute.args[1] - The State, Dependencies, Params, Module, and Type object
+ * 
+ * @returns {Plugin} A plugin that runs during command execution flow
+ * 
+ * @example
+ * // Plugin to restrict command to specific guild
+ * export const inGuild = (guildId: string) => {
+ *   return CommandControlPlugin((ctx, sdt) => {
+ *     if(ctx.guild.id !== guildId) {
+ *       return controller.stop();
+ *     }
+ *     return controller.next();
+ *   });
+ * };
+ * 
+ * @example
+ * // Plugins passing state through the chain
+ * const plugin1 = CommandControlPlugin((ctx, sdt) => {
+ *   return controller.next({ 'plugin1/data': 'from plugin1' });
+ * });
+ * 
+ * const plugin2 = CommandControlPlugin((ctx, sdt) => {
+ *   return controller.next({ 'plugin2/data': ctx.user.id });
+ * });
+ * 
+ * export default commandModule({
+ *   type: CommandType.Slash,
+ *   plugins: [plugin1, plugin2],
+ *   execute: (ctx, sdt) => {
+ *     console.log(sdt.state); // Access accumulated state
+ *   }
+ * });
+ * 
+ * @remarks
+ * - Control plugins are executed in order when a discord.js event is emitted
+ * - Use controller.next() to continue to next plugin or controller.stop() to halt execution
+ * - State can be passed between plugins using controller.next({ key: value })
+ * - State keys should be namespaced to avoid collisions (e.g., 'plugin-name/key')
+ * - Final accumulated state is passed to the command's execute function
+ * - All plugins must succeed for the command to execute
+ * - Plugins have access to dependencies through the sdt.deps object
+ * - Useful for implementing preconditions, filters, and command preprocessing
  */
 export function CommandControlPlugin<I extends CommandType>(
     execute: (...args: CommandArgs<I>) => PluginResult,

src/handlers/event-utils.ts

@@ -1,219 +1,18 @@
-import type { Interaction, Message, BaseInteraction } from 'discord.js';
+import type { Emitter, Logging } from '../core/interfaces';
-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 { SernError } from '../core/structures/enums'
-import { EMPTY_ERR, Err, Ok, Result,  wrapAsync } from '../core/structures/result';
+import { Ok, wrapAsync} from '../core/structures/result';
-import type { UnpackedDependencies } from '../types/utility';
+import type {  Module } from '../types/core-modules';
-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 { inspect } from 'node:util'
-import { disposeAll } from '../core/ioc';
+import { resultPayload } from '../core/functions'
-import { resultPayload, isAutocomplete, treeSearch, fmt } 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  = {
+        const result = await plugin.execute(...args);
-            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);
         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);
-         }))

src/handlers/interaction.ts

@@ -1,30 +1,58 @@
-import type { Interaction } from 'discord.js';
+import type { Module } from '../types/core-modules'
-import { mergeMap, merge, concatMap, EMPTY } from 'rxjs';
+import {  callPlugins, executeModule } from './event-utils';
-import { createInteractionHandler, executeModule, intoTask, sharedEventStream, filterTap, handleCrash } from './event-utils';
 import { SernError } from '../core/structures/enums'
-import { isAutocomplete, isCommand, isMessageComponent, isModal, resultPayload } from '../core/functions'
+import { createSDT, isAutocomplete, isCommand, isMessageComponent, isModal, resultPayload, treeSearch } from '../core/functions'
-import { UnpackedDependencies } from '../types/utility';
+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
+            '@sern/modules': moduleManager,
-    const interactionStream$ = sharedEventStream<Interaction>(client, 'interactionCreate');
+            '@sern/logger': log,
-    const handle = createInteractionHandler(interactionStream$, deps, defaultPrefix);
+            '@sern/emitter': reporter } = deps
 
-    const interactionHandler$ = merge(handle(isMessageComponent),
+    client.on('interactionCreate', async (event) => {
-                                      handle(isAutocomplete),
+
-                                      handle(isCommand),
+        //returns array of possible ids
-                                      handle(isModal));
+        const possibleIds = Id.reconstruct(event);
-    return interactionHandler$
+
-        .pipe(filterTap(e => emitter.emit('warning', resultPayload('warning', undefined, e))),
+        let modules = possibleIds
-              concatMap(intoTask(module => {
+            .map(({ id, params }) => ({ module: moduleManager.get(id)!, params }))
-                emitter.emit('module.activate', resultPayload('failure', module, SernError.PluginFailure))
+            .filter(({ module }) => module !== undefined);
-              })),
+
-              mergeMap(payload => {
+        if(modules.length == 0) {
-                  if(payload)
+            return;
-                    return executeModule(emitter, payload)
+        }
-                  return EMPTY;
+        const { module, params } = modules.at(0)!;
-              }),
+        let payload;
-              handleCrash(deps, "interaction handling"));
+        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);
+    });
 }

src/handlers/message.ts

@@ -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 { createSDT, fmt, resultPayload } from '../core/functions'
-import { UnpackedDependencies } from '../types/utility';
+import type { UnpackedDependencies } from '../types/utility';
-import type { Emitter } from '../core/interfaces';
+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) {
+function isBotOrNoPrefix(msg: Message, prefix: string) {
-    return (msg: Message): msg is Message => !msg.author.bot && hasPrefix(prefix, msg.content);
+    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 
+export function messageHandler (deps: UnpackedDependencies, defaultPrefix?: string) {
-function (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');
+    client.on('messageCreate', async message => {
-    const handle = createMessageHandler(messageStream$, defaultPrefix, deps);
+        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
+
+        executeModule(emitter, log, payload)
+    })
     
-    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")
-    )
 }

src/handlers/presence.ts

@@ -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>) => {
+const parseConfig = async (conf: Promise<Presence.Result>, setPresence: SetPresence) => {
-    return conf.then(s => {
+    const result = await conf;
-        if('repeat' in s) {
+    
-            const { onRepeat, repeat } = s;
+    if ('repeat' in result) {
-            assert(repeat !== undefined, "repeat option is undefined");
+        const { onRepeat, repeat } = result;
-            assert(onRepeat !== undefined, "onRepeat callback is undefined, but repeat exists");
+        // Validate configuration
-            const src$ = typeof repeat === 'number' 
+        if (repeat === undefined) {
-                ? interval(repeat)
+            throw new Error("repeat option is undefined");
-                : fromEvent(...repeat);
-                return src$.pipe(mergeScan(async (args) => onRepeat(args), s), 
-                                 startWith(s));
         }
-        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) => {
+export const presenceHandler = async (path: string, setPresence: SetPresence) => {
-    const presence = Files
+    const presence = await 
-        .importModule<Presence.Config<(keyof Dependencies)[]>>(path)
+        Files.importModule<Presence.Config<(keyof Dependencies)[]>>(path)
-        .then(({ module }) => {
+             .then(({ module }) => {
-            //fetch services with the order preserved, passing it to the execute fn 
+                //fetch services with the order preserved, passing it to the execute fn 
-            const fetchedServices = Services(...module.inject ?? []);
+                const fetchedServices = Services(...module.inject ?? []);
-            return async () => module.execute(...fetchedServices);
+                return async () => module.execute(...fetchedServices);
-        })
+             })
-    const module$ = from(presence);
+     
-    return module$.pipe(
+    return parseConfig(presence(), setPresence);
-        //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))));
 }

src/handlers/ready.ts

@@ -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) {

src/handlers/user-defined-events.ts

@@ -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) => 
+export default async function(deps: UnpackedDependencies, wrapper: Wrapper) {
-    (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) {
     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)
+    const logger = deps['@sern/logger'], report = deps['@sern/emitter'];
-        .pipe(map(intoDispatcher(deps)),
+    for (const module of eventModules) {
-               mergeAll(), // all eventListeners are turned on
+        let source: Emitter;
-               handleCrash(deps, "event modules"))
+
-            .subscribe();
+        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)
+    }
 }
+

src/index.ts

@@ -38,7 +38,7 @@ export type {
 } from './types/core-plugin';
 
 
-export type { Payload, SernEventsMapping } from './types/utility';
+export type { Payload, SernEventsMapping, Wrapper } from './types/utility';
 
 export {
     commandModule,

src/sern.ts

@@ -1,24 +1,21 @@
 //side effect: global container
 import { useContainerRaw } from '@sern/ioc/global';
+// set asynchronous capturing of errors
+import events from 'node:events'
+events.captureRejections = true;
 
 import callsites from 'callsites';
 import * as  Files from './core/module-loading';
-import { merge } from 'rxjs';
 import eventsHandler from './handlers/user-defined-events';
 import ready  from './handlers/ready';
-import messageHandler from './handlers/message';
+import { interactionHandler } from './handlers/interaction';
-import interactionHandler from './handlers/interaction';
+import { messageHandler } from './handlers/message'
 import { presenceHandler } from './handlers/presence';
-import { UnpackedDependencies } from './types/utility';
+import { UnpackedDependencies, Wrapper } from './types/utility';
 import type { Presence} from './core/presences';
 import { registerTasks } from './handlers/tasks';
 
-interface Wrapper {
+
-    commands: string;
-    defaultPrefix?: string;
-    events?: string;
-    tasks?: string;
-}
 /**
  * @since 1.0.0
  * @param maybeWrapper Options to pass into sern.
@@ -37,7 +34,7 @@ export function init(maybeWrapper: Wrapper = { commands: "./dist/commands" }) {
     const deps = useContainerRaw().deps<UnpackedDependencies>();
     
     if (maybeWrapper.events !== undefined) {
-        eventsHandler(deps, maybeWrapper.events)
+        eventsHandler(deps, maybeWrapper)
             .then(() => {
                 deps['@sern/logger']?.info({ message: "Events registered" });
             });
@@ -56,7 +53,7 @@ export function init(maybeWrapper: Wrapper = { commands: "./dist/commands" }) {
                 const setPresence = async (p: Presence.Result) => {
                     return deps['@sern/client'].user?.setPresence(p);
                 }
-                presenceHandler(presencePath.path, setPresence).subscribe();
+                presenceHandler(presencePath.path, setPresence);
             }
             if(maybeWrapper.tasks) {
                 registerTasks(maybeWrapper.tasks, deps);
@@ -64,8 +61,9 @@ export function init(maybeWrapper: Wrapper = { commands: "./dist/commands" }) {
         })
         .catch(err => { throw err });
 
-    const messages$ = messageHandler(deps, maybeWrapper.defaultPrefix);
+    //const messages$ = messageHandler(deps, maybeWrapper.defaultPrefix);
-    const interactions$ = interactionHandler(deps, maybeWrapper.defaultPrefix);
+    interactionHandler(deps, maybeWrapper.defaultPrefix);
+    messageHandler(deps, maybeWrapper.defaultPrefix)
     // listening to the message stream and interaction stream
-    merge(messages$, interactions$).subscribe();
+    //merge(messages$, interactions$).subscribe();
 }

src/types/utility.ts

@@ -1,11 +1,9 @@
 import type { InteractionReplyOptions, MessageReplyOptions } from 'discord.js';
 import type { Module } from './core-modules';
-import type { Result } from '../core/structures/result';
 
 export type Awaitable<T> = PromiseLike<T> | T;
 export type Dictionary = Record<string, unknown>
 
-export type VoidResult = Result<void, void>;
 export type AnyFunction = (...args: any[]) => unknown;
 
 export interface SernEventsMapping {
@@ -26,3 +24,11 @@ export type UnpackedDependencies = {
     [K in keyof Dependencies]: UnpackFunction<Dependencies[K]>
 }
 export type ReplyOptions = string | Omit<InteractionReplyOptions, 'fetchReply'> | MessageReplyOptions;
+
+
+export interface Wrapper {
+    commands: string;
+    defaultPrefix?: string;
+    events?: string;
+    tasks?: string;
+}

test/core/presence.test.ts

@@ -1,6 +1,7 @@
-import { describe, expect, it, vi } from 'vitest';
+import { beforeEach, describe, expect, it, vi } from 'vitest';
 import { Presence } from '../../src';
-
+import * as Files from '../../src/core/module-loading'
+import { presenceHandler } from '../../src/handlers/presence'
 
 // Example test suite for the module function
 describe('module function', () => {
@@ -54,4 +55,38 @@ describe('of function', () => {
       activities: [{ name: 'Another Test Activity' }],
     });
   });
+
+
+})
+
+
+describe('Presence module execution', () => {
+    const mockExecuteResult = Presence.of({
+        status: 'online',
+    }).once();
+  
+    const mockModule = Presence.module({ 
+        inject: [ '@sern/client'],
+        execute: vi.fn().mockReturnValue(mockExecuteResult)
+    })
+    beforeEach(() => {
+        vi.clearAllMocks();
+        // Mock Files.importModule
+        vi.spyOn(Files, 'importModule').mockResolvedValue({
+          module: mockModule
+        });
+
+
+        
+    }); 
+    it('should set presence once.', async () => {
+        const setPresenceMock = vi.fn();
+        const mockPath = '/path/to/presence/config';
+    
+        await presenceHandler(mockPath, setPresenceMock);
+
+        expect(Files.importModule).toHaveBeenCalledWith(mockPath);
+        expect(setPresenceMock).toHaveBeenCalledOnce();
+    })
+
 })

test/handlers.test.ts

@@ -1,13 +1,11 @@
 //@ts-nocheck
 import { beforeEach, describe, expect, it, test } from 'vitest';
-import { callInitPlugins, eventDispatcher } from '../src/handlers/event-utils';
+import { callInitPlugins } from '../src/handlers/event-utils';
 
 import { Client } from 'discord.js'
 import { faker } from '@faker-js/faker';
-import { Module } from '../src/types/core-modules';
-import { Processed } from '../src/types/core-modules';
 import { EventEmitter } from 'events';
-import { CommandControlPlugin, CommandInitPlugin, CommandType, controller } from '../src';
+import { CommandControlPlugin, CommandType, controller } from '../src';
 import { createRandomModule, createRandomInitPlugin } from './setup/util';
 
 
@@ -19,23 +17,6 @@ function mockDeps() {
    }
 }
  
-describe('eventDispatcher standard', () => {
-    let m: Processed<Module>;
-    let ee: EventEmitter;
-    beforeEach(() => {
-        ee = new EventEmitter();
-        m = createRandomModule();
-    });
-
-    it('should throw', () => {
-        expect(() => eventDispatcher(mockDeps(), m,  'not event emitter')).toThrowError();
-    });
-
-    it("Shouldn't throw", () => {
-        expect(() => eventDispatcher(mockDeps(), m, ee)).not.toThrowError();
-    });
-});
-
 describe('calling init plugins', async () => {
    let deps;
    beforeEach(() => {

test/setup/setup-tests.ts

@@ -1,4 +1,6 @@
 import { vi } from 'vitest'
+import { makeDependencies } from '../../src';
+import { Client } from 'discord.js';
 
 vi.mock('discord.js', async (importOriginal) => {
     const mod = await importOriginal()
@@ -44,3 +46,9 @@ vi.mock('discord.js', async (importOriginal) => {
         ChatInputCommandInteraction: vi.fn()
     };
 });
+
+await makeDependencies(({ add }) => {
+    add('@sern/client', {  })
+})
+
+