feat(engine): add message queue, triggers, and orchestrators
New @tts/engine package: the pure message layer that drives bgm games. MessageQueue snapshots and drains on tick; TriggerRegistry matches by type/id with runtime enable/disable; runOrchestrator suspends on wait until a matching message is processed. Engine ties them together behind the CommandHost contract that @tts/tabletop will implement.
This commit is contained in:
@@ -0,0 +1,23 @@
|
||||
{
|
||||
"name": "@tts/engine",
|
||||
"version": "0.0.0",
|
||||
"private": true,
|
||||
"type": "module",
|
||||
"main": "./dist/index.js",
|
||||
"types": "./dist/index.d.ts",
|
||||
"exports": {
|
||||
".": {
|
||||
"types": "./dist/index.d.ts",
|
||||
"import": "./dist/index.js"
|
||||
}
|
||||
},
|
||||
"scripts": {
|
||||
"build": "tsc -p tsconfig.json",
|
||||
"typecheck": "tsc -p tsconfig.json --noEmit",
|
||||
"test": "vitest run",
|
||||
"lint": "echo \"no lint configured\""
|
||||
},
|
||||
"devDependencies": {
|
||||
"typescript": "^5.7.2"
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,76 @@
|
||||
import { describe, expect, it } from 'vitest';
|
||||
import { Engine, type CommandHost, type Message } from './engine.js';
|
||||
|
||||
/** A minimal command host that runs a `move` and emits `move:done`. */
|
||||
function moveHost(): CommandHost & { moves: Message[] } {
|
||||
const handlers = new Map<string, (msg: Message) => void>();
|
||||
const moves: Message[] = [];
|
||||
return {
|
||||
moves,
|
||||
on(type, handler) {
|
||||
handlers.set(type, handler);
|
||||
return () => handlers.delete(type);
|
||||
},
|
||||
dispatch(msg) {
|
||||
if (msg.type === 'move') {
|
||||
moves.push(msg);
|
||||
// The command's completion is itself a message.
|
||||
this.dispatch({ type: 'move:done', data: msg.data });
|
||||
} else if (msg.type === 'move:done') {
|
||||
handlers.get('move:done')?.(msg);
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
describe('Engine', () => {
|
||||
it('dispatches a message to the command host on tick', () => {
|
||||
const host = moveHost();
|
||||
const engine = new Engine(host);
|
||||
engine.enqueue({ type: 'move', data: { part: 'a', to: '/grid/5/5' } });
|
||||
engine.tick();
|
||||
expect(host.moves).toHaveLength(1);
|
||||
});
|
||||
|
||||
it('a trigger reacts to a message and emits on the next tick', () => {
|
||||
const host = moveHost();
|
||||
const engine = new Engine(host);
|
||||
engine.registerTrigger({
|
||||
type: 'tap',
|
||||
id: 'draw',
|
||||
match: { part: 'carcassonne:tile#a' },
|
||||
emit: [{ type: 'move', data: { part: 'carcassonne:tile#a', to: '/grid/5/5' } }],
|
||||
});
|
||||
|
||||
engine.enqueue({ type: 'tap', data: { part: 'carcassonne:tile#a', trigger: 'draw' } });
|
||||
engine.tick(); // tap processed; move emitted to the next tick
|
||||
expect(host.moves).toHaveLength(0);
|
||||
engine.tick(); // move runs
|
||||
expect(host.moves).toHaveLength(1);
|
||||
});
|
||||
|
||||
it('an orchestrator awaits a matching message and resumes on tick', async () => {
|
||||
const host = moveHost();
|
||||
const engine = new Engine(host);
|
||||
const log: string[] = [];
|
||||
|
||||
const done = engine.runOrchestrator(async (ctx) => {
|
||||
log.push('start');
|
||||
const tap = await ctx.wait((m) => m.type === 'tap');
|
||||
log.push(`tap:${(tap.data as { part: string }).part}`);
|
||||
ctx.emit({ type: 'move', data: { part: 'a', to: '/grid/5/5' } });
|
||||
});
|
||||
|
||||
// Nothing enqueued yet — the orchestrator is suspended.
|
||||
engine.tick();
|
||||
expect(log).toEqual(['start']);
|
||||
|
||||
engine.enqueue({ type: 'tap', data: { part: 'carcassonne:tile#a' } });
|
||||
engine.tick();
|
||||
await done;
|
||||
expect(log).toEqual(['start', 'tap:carcassonne:tile#a']);
|
||||
// The move emitted by the orchestrator runs on the next tick.
|
||||
engine.tick();
|
||||
expect(host.moves).toHaveLength(1);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,92 @@
|
||||
/**
|
||||
* The engine: the message bus that ties the queue, triggers, orchestrators,
|
||||
* and the command host together.
|
||||
*
|
||||
* The engine is pure — no r3f, no React, no store. It defines the contract;
|
||||
* `@tts/tabletop` implements the `CommandHost` with the built-in commands that
|
||||
* mutate the tabletop store and drive the render layer. The engine never
|
||||
* imports tabletop.
|
||||
*/
|
||||
import { MessageQueue, type Message } from './message.js';
|
||||
import { TriggerRegistry, type Trigger } from './trigger.js';
|
||||
import { runOrchestrator, type Orchestrator } from './orchestrator.js';
|
||||
|
||||
/**
|
||||
* The command host: how a command handler registers with the bus. `@tts/tabletop`
|
||||
* implements this with the built-in commands (`move`, `focus`, `caption`, ...).
|
||||
* The engine dispatches each message to the host's registered handler for its
|
||||
* type; the host runs the command (the single mutation path) and emits the
|
||||
* `type:done` result message.
|
||||
*/
|
||||
export interface CommandHost {
|
||||
/** Register a handler for a message type. Returns an unsubscribe. */
|
||||
on(type: string, handler: (msg: Message) => void): () => void;
|
||||
/** Dispatch a message to the registered handler for its type, if any. */
|
||||
dispatch(msg: Message): void;
|
||||
}
|
||||
|
||||
/**
|
||||
* The engine. `enqueue` adds a message; `tick` drains the queue and processes
|
||||
* it through the command host and triggers. Orchestrators run against the same
|
||||
* queue and suspend on `wait` until a matching message is processed.
|
||||
*/
|
||||
export class Engine {
|
||||
private queue = new MessageQueue();
|
||||
private triggers = new TriggerRegistry();
|
||||
private host: CommandHost;
|
||||
|
||||
constructor(host: CommandHost) {
|
||||
this.host = host;
|
||||
// Every message goes to the command host first (it runs the command), then
|
||||
// to triggers (they react to the message, including the `type:done` the
|
||||
// host emits). Emissions from either land on the next tick.
|
||||
this.queue.on((msg) => {
|
||||
this.host.dispatch(msg);
|
||||
for (const t of this.triggers.match(msg)) {
|
||||
for (const emit of t.emit) this.queue.enqueue(emit);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
enqueue(msg: Message): void {
|
||||
this.queue.enqueue(msg);
|
||||
}
|
||||
|
||||
/** Drain the queue and process the snapshot. Returns the processed count. */
|
||||
tick(): number {
|
||||
return this.queue.tick();
|
||||
}
|
||||
|
||||
registerTrigger(t: Trigger): void {
|
||||
this.triggers.register(t);
|
||||
}
|
||||
|
||||
unregisterTrigger(t: Trigger): void {
|
||||
this.triggers.unregister(t);
|
||||
}
|
||||
|
||||
enableTrigger(type: string, id?: string): void {
|
||||
this.triggers.enable(type, id);
|
||||
}
|
||||
|
||||
disableTrigger(type: string, id?: string): void {
|
||||
this.triggers.disable(type, id);
|
||||
}
|
||||
|
||||
/** Run an orchestrator against this engine's queue. */
|
||||
runOrchestrator(o: Orchestrator): Promise<void> {
|
||||
return runOrchestrator(
|
||||
o,
|
||||
(msg) => this.queue.enqueue(msg),
|
||||
(handler) => this.queue.on(handler),
|
||||
(type, id) => this.triggers.enable(type, id),
|
||||
(type, id) => this.triggers.disable(type, id),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
export type { Message, MessageHandler } from './message.js';
|
||||
export type { Trigger } from './trigger.js';
|
||||
export { triggerMatches, TriggerRegistry } from './trigger.js';
|
||||
export type { Orchestrator, OrchestratorContext } from './orchestrator.js';
|
||||
export { runOrchestrator } from './orchestrator.js';
|
||||
@@ -0,0 +1,4 @@
|
||||
export { MessageQueue, type Message, type MessageHandler } from './message.js';
|
||||
export { TriggerRegistry, triggerMatches, type Trigger } from './trigger.js';
|
||||
export { runOrchestrator, type Orchestrator, type OrchestratorContext } from './orchestrator.js';
|
||||
export { Engine, type CommandHost } from './engine.js';
|
||||
@@ -0,0 +1,43 @@
|
||||
import { describe, expect, it } from 'vitest';
|
||||
import { MessageQueue } from './message.js';
|
||||
|
||||
describe('MessageQueue', () => {
|
||||
it('processes messages in FIFO order on tick', () => {
|
||||
const q = new MessageQueue();
|
||||
const seen: string[] = [];
|
||||
q.on((m) => seen.push(m.type));
|
||||
q.enqueue({ type: 'a' });
|
||||
q.enqueue({ type: 'b' });
|
||||
expect(q.size).toBe(2);
|
||||
expect(q.tick()).toBe(2);
|
||||
expect(seen).toEqual(['a', 'b']);
|
||||
expect(q.size).toBe(0);
|
||||
});
|
||||
|
||||
it('snapshots and drains: emissions during a drain go to the next tick', () => {
|
||||
const q = new MessageQueue();
|
||||
const seen: string[] = [];
|
||||
q.on((m) => {
|
||||
seen.push(m.type);
|
||||
if (m.type === 'a') q.enqueue({ type: 'b' });
|
||||
});
|
||||
q.enqueue({ type: 'a' });
|
||||
// The 'b' emitted during the drain must NOT be processed in the same tick.
|
||||
expect(q.tick()).toBe(1);
|
||||
expect(seen).toEqual(['a']);
|
||||
expect(q.tick()).toBe(1);
|
||||
expect(seen).toEqual(['a', 'b']);
|
||||
});
|
||||
|
||||
it('unsubscribes a handler', () => {
|
||||
const q = new MessageQueue();
|
||||
const seen: string[] = [];
|
||||
const off = q.on((m) => seen.push(m.type));
|
||||
q.enqueue({ type: 'a' });
|
||||
q.tick();
|
||||
off();
|
||||
q.enqueue({ type: 'b' });
|
||||
q.tick();
|
||||
expect(seen).toEqual(['a']);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,64 @@
|
||||
/**
|
||||
* The message model and the queue that serializes it.
|
||||
*
|
||||
* A message is both an event (something happened) and an intent (something
|
||||
* should happen). It is identified by `type` (and optionally `id`), matching
|
||||
* the format's `type#id` convention. A `move` message both runs the move
|
||||
* command and is observable as an event; the command's completion is itself a
|
||||
* message (`move:done`), which is what triggers match and orchestrators await.
|
||||
*
|
||||
* Messages are not processed inline. They are enqueued and handled on the next
|
||||
* `tick()`. This kills reentrancy (a handler cannot cause unbounded
|
||||
* recursion), gives a natural debounce, and makes the whole system a
|
||||
* deterministic frame. The engine is pure — it has no render loop — so the
|
||||
* host calls `tick()` (a `useFrame` in `@tts/tabletop`, manually in tests).
|
||||
*/
|
||||
export interface Message {
|
||||
type: string;
|
||||
/** Command-specific payload. */
|
||||
data?: unknown;
|
||||
/** Optional identity, for matching and dedup. */
|
||||
id?: string;
|
||||
}
|
||||
|
||||
/** A handler consumes a message and may emit new ones. */
|
||||
export type MessageHandler = (msg: Message) => void;
|
||||
|
||||
/**
|
||||
* A message queue. `enqueue` adds a message to the pending set; `tick` drains
|
||||
* the snapshot and processes it. Messages emitted during a drain go to the
|
||||
* next tick (snapshot-and-drain), so a handler can never re-enter mid-drain.
|
||||
*/
|
||||
export class MessageQueue {
|
||||
private pending: Message[] = [];
|
||||
private handlers: MessageHandler[] = [];
|
||||
|
||||
/** Register a handler for every message. Returns an unsubscribe. */
|
||||
on(handler: MessageHandler): () => void {
|
||||
this.handlers.push(handler);
|
||||
return () => {
|
||||
this.handlers = this.handlers.filter((h) => h !== handler);
|
||||
};
|
||||
}
|
||||
|
||||
enqueue(msg: Message): void {
|
||||
this.pending.push(msg);
|
||||
}
|
||||
|
||||
/** Drain the current snapshot and process it. Returns the processed count. */
|
||||
tick(): number {
|
||||
const batch = this.pending;
|
||||
this.pending = [];
|
||||
for (const msg of batch) {
|
||||
for (const handler of this.handlers) {
|
||||
handler(msg);
|
||||
}
|
||||
}
|
||||
return batch.length;
|
||||
}
|
||||
|
||||
/** The number of messages waiting to be processed. */
|
||||
get size(): number {
|
||||
return this.pending.length;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,64 @@
|
||||
/**
|
||||
* Orchestrators — imperative async flow.
|
||||
*
|
||||
* An orchestrator is the code counterpart to a trigger: an async function that
|
||||
* emits messages and awaits matching ones. It is a proper TS module, declared
|
||||
* per folder as `main.ts`, unique per folder like `package.yaml`.
|
||||
*
|
||||
* An orchestrator is a long-running command: it awaits events instead of
|
||||
* resolving immediately, so it inherits the run-context machinery (supersede
|
||||
* groups, cancellation, tap subscription) for free. Trigger control lives here
|
||||
* — declaration is data, activation is code.
|
||||
*/
|
||||
import type { Message } from './message.js';
|
||||
|
||||
/** The context an orchestrator runs against. */
|
||||
export interface OrchestratorContext {
|
||||
/** Emit a message onto the queue. */
|
||||
emit(msg: Message): void;
|
||||
/** Await the next message matching `pred`. */
|
||||
wait(pred: (msg: Message) => boolean): Promise<Message>;
|
||||
/** Enable/disable a trigger by `type#id`. */
|
||||
enableTrigger(type: string, id?: string): void;
|
||||
disableTrigger(type: string, id?: string): void;
|
||||
}
|
||||
|
||||
/** An orchestrator: an async function that emits and awaits messages. */
|
||||
export type Orchestrator = (ctx: OrchestratorContext) => Promise<void>;
|
||||
|
||||
/**
|
||||
* Run an orchestrator against a queue. `emit` enqueues; `wait` suspends until
|
||||
* a matching message is processed during a `tick()`. Returns a promise that
|
||||
* resolves when the orchestrator completes.
|
||||
*/
|
||||
export function runOrchestrator(
|
||||
orchestrator: Orchestrator,
|
||||
emit: (msg: Message) => void,
|
||||
on: (handler: (msg: Message) => void) => () => void,
|
||||
enableTrigger: (type: string, id?: string) => void,
|
||||
disableTrigger: (type: string, id?: string) => void,
|
||||
): Promise<void> {
|
||||
const pending: Array<{ pred: (msg: Message) => boolean; resolve: (m: Message) => void }> = [];
|
||||
const unsubscribe = on((msg) => {
|
||||
for (let i = 0; i < pending.length; i++) {
|
||||
const p = pending[i]!;
|
||||
if (p.pred(msg)) {
|
||||
pending.splice(i, 1);
|
||||
p.resolve(msg);
|
||||
break;
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
const ctx: OrchestratorContext = {
|
||||
emit,
|
||||
wait: (pred) =>
|
||||
new Promise<Message>((resolve) => {
|
||||
pending.push({ pred, resolve });
|
||||
}),
|
||||
enableTrigger,
|
||||
disableTrigger,
|
||||
};
|
||||
|
||||
return orchestrator(ctx).finally(unsubscribe);
|
||||
}
|
||||
@@ -0,0 +1,59 @@
|
||||
import { describe, expect, it } from 'vitest';
|
||||
import { TriggerRegistry, triggerMatches, type Trigger } from './trigger.js';
|
||||
|
||||
const tap: Trigger = {
|
||||
type: 'tap',
|
||||
id: 'draw',
|
||||
match: { part: 'carcassonne:tile#a', trigger: 'draw' },
|
||||
emit: [{ type: 'move', data: { part: 'carcassonne:tile#a', to: '/grid/5/5' } }],
|
||||
};
|
||||
|
||||
describe('triggerMatches', () => {
|
||||
it('matches on type and every match param', () => {
|
||||
expect(
|
||||
triggerMatches(tap, { type: 'tap', data: { part: 'carcassonne:tile#a', trigger: 'draw' } }),
|
||||
).toBe(true);
|
||||
});
|
||||
|
||||
it('rejects a different type', () => {
|
||||
expect(triggerMatches(tap, { type: 'focus' })).toBe(false);
|
||||
});
|
||||
|
||||
it('rejects a mismatched param', () => {
|
||||
expect(
|
||||
triggerMatches(tap, { type: 'tap', data: { part: 'carcassonne:tile#b', trigger: 'draw' } }),
|
||||
).toBe(false);
|
||||
});
|
||||
|
||||
it('matches any message of the type when there is no match block', () => {
|
||||
const any = { type: 'focus', emit: [] };
|
||||
expect(triggerMatches(any, { type: 'focus', data: { path: '/deck' } })).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
describe('TriggerRegistry', () => {
|
||||
it('registers and matches enabled triggers', () => {
|
||||
const reg = new TriggerRegistry();
|
||||
reg.register(tap);
|
||||
expect(reg.match({ type: 'tap', data: { part: 'carcassonne:tile#a', trigger: 'draw' } })).toEqual([
|
||||
tap,
|
||||
]);
|
||||
});
|
||||
|
||||
it('collision-checks duplicate type#id', () => {
|
||||
const reg = new TriggerRegistry();
|
||||
reg.register(tap);
|
||||
expect(() => reg.register({ ...tap })).toThrow(/Duplicate trigger: tap#draw/);
|
||||
});
|
||||
|
||||
it('disable/enable toggles a trigger at runtime', () => {
|
||||
const reg = new TriggerRegistry();
|
||||
reg.register(tap);
|
||||
reg.disable('tap', 'draw');
|
||||
expect(reg.match({ type: 'tap', data: { part: 'carcassonne:tile#a', trigger: 'draw' } })).toEqual([]);
|
||||
reg.enable('tap', 'draw');
|
||||
expect(reg.match({ type: 'tap', data: { part: 'carcassonne:tile#a', trigger: 'draw' } })).toEqual([
|
||||
tap,
|
||||
]);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,77 @@
|
||||
/**
|
||||
* Triggers — declarative reactive glue.
|
||||
*
|
||||
* A trigger matches a message by `type` and named params, and emits messages
|
||||
* in response. It is declared as data, keyed by `role+type+id` like other
|
||||
* defs, and collision-checked the same way. `match` binds named params from
|
||||
* the payload (like a route's candidates); `emit` uses the command-id-as-key
|
||||
* convention. Multiple triggers can match the same message — both fire.
|
||||
*/
|
||||
import type { Message } from './message.js';
|
||||
|
||||
export interface Trigger {
|
||||
/** The message type this trigger matches. */
|
||||
type: string;
|
||||
/** Optional identity, for runtime enable/disable and collision checks. */
|
||||
id?: string;
|
||||
/** Named params that must equal the corresponding fields in `msg.data`. */
|
||||
match?: Record<string, unknown>;
|
||||
/** Messages to emit when the trigger matches. */
|
||||
emit: Message[];
|
||||
}
|
||||
|
||||
/** A trigger matches when its `type` and every `match` param line up. */
|
||||
export function triggerMatches(t: Trigger, msg: Message): boolean {
|
||||
if (t.type !== msg.type) return false;
|
||||
if (!t.match) return true;
|
||||
const data = msg.data as Record<string, unknown> | undefined;
|
||||
if (!data) return false;
|
||||
return Object.entries(t.match).every(([k, v]) => data[k] === v);
|
||||
}
|
||||
|
||||
/**
|
||||
* A registry of triggers, keyed by `type#id`. `register` collision-checks the
|
||||
* key; `enable`/`disable` toggle a trigger at runtime (an orchestrator's
|
||||
* "no more placements this turn" control). `match` returns every enabled
|
||||
* trigger that matches a message.
|
||||
*/
|
||||
export class TriggerRegistry {
|
||||
private triggers = new Map<string, Trigger>();
|
||||
private enabled = new Set<string>();
|
||||
|
||||
register(t: Trigger): void {
|
||||
const key = triggerKey(t);
|
||||
if (this.triggers.has(key)) {
|
||||
throw new Error(`Duplicate trigger: ${key}`);
|
||||
}
|
||||
this.triggers.set(key, t);
|
||||
this.enabled.add(key);
|
||||
}
|
||||
|
||||
unregister(t: Trigger): void {
|
||||
const key = triggerKey(t);
|
||||
this.triggers.delete(key);
|
||||
this.enabled.delete(key);
|
||||
}
|
||||
|
||||
enable(type: string, id?: string): void {
|
||||
this.enabled.add(triggerKey({ type, id }));
|
||||
}
|
||||
|
||||
disable(type: string, id?: string): void {
|
||||
this.enabled.delete(triggerKey({ type, id }));
|
||||
}
|
||||
|
||||
/** Every enabled trigger matching `msg`. */
|
||||
match(msg: Message): Trigger[] {
|
||||
const out: Trigger[] = [];
|
||||
for (const [key, t] of this.triggers) {
|
||||
if (this.enabled.has(key) && triggerMatches(t, msg)) out.push(t);
|
||||
}
|
||||
return out;
|
||||
}
|
||||
}
|
||||
|
||||
function triggerKey(t: Pick<Trigger, 'type' | 'id'>): string {
|
||||
return t.id ? `${t.type}#${t.id}` : t.type;
|
||||
}
|
||||
@@ -0,0 +1,9 @@
|
||||
{
|
||||
"extends": "../../tsconfig.base.json",
|
||||
"compilerOptions": {
|
||||
"outDir": "./dist",
|
||||
"rootDir": "./src",
|
||||
"lib": ["ES2022", "DOM"]
|
||||
},
|
||||
"include": ["src"]
|
||||
}
|
||||
Generated
+6
@@ -183,6 +183,12 @@ importers:
|
||||
specifier: ^4.1.10
|
||||
version: 4.1.10(@types/node@22.20.1)(vite@8.2.1(@types/node@22.20.1)(esbuild@0.28.1)(jiti@2.7.0)(tsx@4.23.11)(yaml@2.9.0))
|
||||
|
||||
packages/engine:
|
||||
devDependencies:
|
||||
typescript:
|
||||
specifier: ^5.7.2
|
||||
version: 5.9.3
|
||||
|
||||
packages/extract:
|
||||
dependencies:
|
||||
'@tts/shared':
|
||||
|
||||
Reference in New Issue
Block a user