core/src/js/sw/sdk/pipe.js

// ==========================================
// Switchboard Core — SDK: Pipe/Filter Pipeline
// ==========================================
// Three-stage pipeline: pre-send, stream, render.
// Verbatim port of switchboard-sdk.js pipe code.
//
// Factory: createPipe()
// ==========================================

/**
 * Create the pipe/filter pipeline.
 * @returns {object} pipe
 */
export function createPipe() {

    const _chains = {
        pre:    [],  // { priority, fn, scope, source, stats }
        stream: [],
        render: [],
    };

    /**
     * Register a filter into a pipe stage.
     */
    function _registerFilter(stage, priority, fn, opts) {
        if (typeof fn !== 'function') {
            console.error(`[sw.pipe] ${stage}: filter must be a function`);
            return;
        }
        const chain = _chains[stage];
        if (!chain) {
            console.error(`[sw.pipe] ${stage}: unknown stage`);
            return;
        }
        const _opts = opts || {};
        const entry = {
            priority,
            fn,
            scope:  _opts.scope || null,
            source: _opts.source || _inferSource(),
            stats:  { calls: 0, totalMs: 0, errors: 0 },
        };

        const dup = chain.find(e => e.source === entry.source && e.priority === entry.priority);
        if (dup) {
            console.warn(`[sw.pipe] ${stage}: duplicate registration (source=${entry.source}, priority=${priority})`);
        }

        chain.push(entry);
        chain.sort((a, b) => a.priority - b.priority);
    }

    /**
     * Infer source label from call stack.
     */
    function _inferSource() {
        try {
            const stack = new Error().stack || '';
            const extMatch = stack.match(/extensions\/([^/]+)\//);
            if (extMatch) return extMatch[1];
        } catch (_) { /* best effort */ }
        return 'anonymous';
    }

    /**
     * Run a filter chain against a context.
     * Sync-only — filters must not return Promises.
     */
    function _runChain(stage, ctx) {
        const chain = _chains[stage];
        if (!chain || chain.length === 0) return ctx;

        const channelType = ctx.channel?.type || null;

        for (const entry of chain) {
            if (entry.scope?.channelType) {
                if (!channelType || !entry.scope.channelType.includes(channelType)) {
                    continue;
                }
            }

            const t0 = performance.now();
            try {
                const result = entry.fn(ctx);
                entry.stats.calls++;
                entry.stats.totalMs += performance.now() - t0;

                if (result instanceof Promise) {
                    console.error(`[sw.pipe] ${stage}: filter '${entry.source}' returned Promise (async not supported)`);
                    entry.stats.errors++;
                    continue;
                }

                if (result === null || result === undefined) {
                    return null; // halt chain
                }
                ctx = result;
            } catch (e) {
                entry.stats.calls++;
                entry.stats.errors++;
                entry.stats.totalMs += performance.now() - t0;
                console.error(`[sw.pipe] ${stage} filter '${entry.source}' (p=${entry.priority}) threw:`, e);
                // Continue chain — error isolation
            }
        }
        return ctx;
    }

    // ── Public API ──────────────────────────────

    return {
        pre(priority, fn, opts)    { _registerFilter('pre', priority, fn, opts); },
        stream(priority, fn, opts) { _registerFilter('stream', priority, fn, opts); },
        render(priority, fn, opts) { _registerFilter('render', priority, fn, opts); },

        list() {
            const result = {};
            for (const [stage, chain] of Object.entries(_chains)) {
                result[stage] = chain.map(e => ({
                    priority: e.priority,
                    source:   e.source,
                    scope:    e.scope,
                    calls:    e.stats.calls,
                    avgMs:    e.stats.calls > 0
                        ? Math.round((e.stats.totalMs / e.stats.calls) * 100) / 100
                        : 0,
                    errors:   e.stats.errors,
                }));
            }
            return result;
        },

        _runPre(ctx)    { return _runChain('pre', ctx); },
        _runStream(ctx) { return _runChain('stream', ctx); },
        _runRender(ctx) { return _runChain('render', ctx); },
    };
}