// crm-engine/reactive-bus.js v3.1.0
// EventBus: middleware pipeline, typed events,
// time-travel replay, priority handlers
class EventBus {
#handlers = new Map();
#middleware = [];
#history = [];
#max = 500;
#listeners = new WeakMap();
/** Добавить middleware в pipeline */
use(fn) {
this.#middleware.push(fn);
return this;
}
/** Подписка; возвращает функцию отписки */
on(event, handler, {
once = false,
priority = 0,
} = {}) {
if (!this.#handlers.has(event))
this.#handlers.set(event, []);
const entry = { handler, priority, once };
this.#handlers.get(event).push(entry);
this.#handlers.get(event)
.sort((a, b) => b.priority - a.priority);
return () => this.#off(event, entry);
}
/** Emit через middleware → handlers */
async emit(event, payload) {
const ctx = {
event, payload,
ts: Date.now(),
meta: {},
cancelled: false,
};
const pipe = [
...this.#middleware,
c => this.#dispatch(c),
];
await this.#run(pipe, 0, ctx);
if (!ctx.cancelled) {
this.#history.push(Object.freeze({ ...ctx }));
if (this.#history.length > this.#max)
this.#history.shift();
}
return ctx;
}
async #run(pipe, i, ctx) {
if (i >= pipe.length || ctx.cancelled) return;
await pipe[i](ctx, () => this.#run(pipe, i + 1, ctx));
}
async #dispatch({ event, payload }) {
const entries = this.#handlers.get(event) ?? [];
await Promise.allSettled(
entries.map(e => Promise.resolve(e.handler(payload)))
);
entries.filter(e => e.once)
.forEach(e => this.#off(event, e));
}
#off(event, entry) {
const list = this.#handlers.get(event);
const i = list?.indexOf(entry) ?? -1;
if (i !== -1) list.splice(i, 1);
}
get size() {
return [...this.#handlers.values()]
.reduce((n, arr) => n + arr.length, 0);
}
async replay(fromTs = 0) {
for (const snap of this.#history) {
if (snap.ts < fromTs) continue;
await this.emit(snap.event, snap.payload);
}
}
}
// ── Инициализация ─────────────────────────────────
const bus = new EventBus()
.use(async (ctx, next) => { // logger
const t = performance.now();
await next();
console.log(`[${ctx.event}] +${(performance.now()-t).toFixed(1)}ms`);
})
.use(async (ctx, next) => { // auth guard
if (ctx.event.startsWith('admin:') && !ctx.meta.role)
return void (ctx.cancelled = true);
await next();
});
// ── Обработчики событий ────────────────────────────
bus.on('order:created', async ({ id, items, userId }) => {
const total = items.reduce(
(sum, { price, qty }) => sum + price * qty, 0
);
await db.transaction(async trx => {
await trx('orders').insert({
id, userId, total, status: 'pending'
});
await trx('inventory').decrement(
Object.fromEntries(items.map(i => [i.sku, i.qty]))
);
});
await Promise.all([
bus.emit('invoice:generate', { orderId: id, total }),
bus.emit('notify:user', {
userId, type: 'order_confirm', ref: id
}),
]);
});
bus.on('invoice:generate', async ({ orderId, total }) => {
const pdf = await renderer.render('invoice', {
orderId, total,
date: new Date().toISOString().slice(0, 10),
});
await storage.put(`invoices/${orderId}.pdf`, pdf);
console.log(`invoice ${orderId} · ${pdf.byteLength >> 10} KB`);
});
bus.on('notify:user', async ({ userId, type, ref }) => {
const user = await db.users.findOne({ id: userId });
await mailer.send({
to: user.email,
subject: TEMPLATES[type].subject(ref),
html: await tpl.render(type, { user, ref }),
});
});
bus.on('order:refund', async ({ orderId, reason }) => {
const order = await db.orders.findOne({ id: orderId });
if (order.status === 'refunded') return;
await db.orders.update({ id: orderId }, {
status: 'refunded',
refundedAt: Date.now(),
reason,
});
await payment.refund(order.paymentId, order.total);
await bus.emit('notify:user', {
userId: order.userId,
type: 'refund_confirm',
ref: orderId,
});
}, { priority: 10 });
// crm-engine/reactive-bus.js v3.1.0
// EventBus: middleware pipeline, typed events,
// time-travel replay, priority handlers
class EventBus {
#handlers = new Map();
#middleware = [];
#history = [];
#max = 500;
#listeners = new WeakMap();
/** Добавить middleware в pipeline */
use(fn) {
this.#middleware.push(fn);
return this;
}
/** Подписка; возвращает функцию отписки */
on(event, handler, {
once = false,
priority = 0,
} = {}) {
if (!this.#handlers.has(event))
this.#handlers.set(event, []);
const entry = { handler, priority, once };
this.#handlers.get(event).push(entry);
this.#handlers.get(event)
.sort((a, b) => b.priority - a.priority);
return () => this.#off(event, entry);
}
/** Emit через middleware → handlers */
async emit(event, payload) {
const ctx = {
event, payload,
ts: Date.now(),
meta: {},
cancelled: false,
};
const pipe = [
...this.#middleware,
c => this.#dispatch(c),
];
await this.#run(pipe, 0, ctx);
if (!ctx.cancelled) {
this.#history.push(Object.freeze({ ...ctx }));
if (this.#history.length > this.#max)
this.#history.shift();
}
return ctx;
}
async #run(pipe, i, ctx) {
if (i >= pipe.length || ctx.cancelled) return;
await pipe[i](ctx, () => this.#run(pipe, i + 1, ctx));
}
async #dispatch({ event, payload }) {
const entries = this.#handlers.get(event) ?? [];
await Promise.allSettled(
entries.map(e => Promise.resolve(e.handler(payload)))
);
entries.filter(e => e.once)
.forEach(e => this.#off(event, e));
}
#off(event, entry) {
const list = this.#handlers.get(event);
const i = list?.indexOf(entry) ?? -1;
if (i !== -1) list.splice(i, 1);
}
get size() {
return [...this.#handlers.values()]
.reduce((n, arr) => n + arr.length, 0);
}
async replay(fromTs = 0) {
for (const snap of this.#history) {
if (snap.ts < fromTs) continue;
await this.emit(snap.event, snap.payload);
}
}
}
// ── Инициализация ─────────────────────────────────
const bus = new EventBus()
.use(async (ctx, next) => { // logger
const t = performance.now();
await next();
console.log(`[${ctx.event}] +${(performance.now()-t).toFixed(1)}ms`);
})
.use(async (ctx, next) => { // auth guard
if (ctx.event.startsWith('admin:') && !ctx.meta.role)
return void (ctx.cancelled = true);
await next();
});
// ── Обработчики событий ────────────────────────────
bus.on('order:created', async ({ id, items, userId }) => {
const total = items.reduce(
(sum, { price, qty }) => sum + price * qty, 0
);
await db.transaction(async trx => {
await trx('orders').insert({
id, userId, total, status: 'pending'
});
await trx('inventory').decrement(
Object.fromEntries(items.map(i => [i.sku, i.qty]))
);
});
await Promise.all([
bus.emit('invoice:generate', { orderId: id, total }),
bus.emit('notify:user', {
userId, type: 'order_confirm', ref: id
}),
]);
});
bus.on('invoice:generate', async ({ orderId, total }) => {
const pdf = await renderer.render('invoice', {
orderId, total,
date: new Date().toISOString().slice(0, 10),
});
await storage.put(`invoices/${orderId}.pdf`, pdf);
console.log(`invoice ${orderId} · ${pdf.byteLength >> 10} KB`);
});
bus.on('notify:user', async ({ userId, type, ref }) => {
const user = await db.users.findOne({ id: userId });
await mailer.send({
to: user.email,
subject: TEMPLATES[type].subject(ref),
html: await tpl.render(type, { user, ref }),
});
});
bus.on('order:refund', async ({ orderId, reason }) => {
const order = await db.orders.findOne({ id: orderId });
if (order.status === 'refunded') return;
await db.orders.update({ id: orderId }, {
status: 'refunded',
refundedAt: Date.now(),
reason,
});
await payment.refund(order.paymentId, order.total);
await bus.emit('notify:user', {
userId: order.userId,
type: 'refund_confirm',
ref: orderId,
});
}, { priority: 10 });
