xiaoxiaofangkuaizhen_emojo/assets/mx/module/startupWorker/StartupWorker.ts

import type {
    StartupEnterGameListener,
    StartupGraph,
    StartupGraphNode,
    StartupOverallListener,
    StartupStateListener,
    StartupUIState,
    StartupWorkContext,
} from './StartupWorkerTypes';

function clamp01(v: number): number {
    if (v < 0) return 0;
    if (v > 1) return 1;
    return v;
}

function sumWeights(nodes: StartupGraphNode[]): number {
    return nodes.reduce((s, n) => s + (n.weight ?? 1), 0);
}

function initialState(): StartupUIState {
    return {
        overall: 0,
        activeId: null,
        activeLabel: null,
        activeProgress: 0,
        scopeId: null,
        scopeLabel: null,
        lastEvent: null,
        phase: 'idle',
        error: null,
    };
}

function checkAborted(signal: AbortSignal | undefined): void {
    if (signal?.aborted) {
        const e = new Error('Startup aborted');
        e.name = 'StartupAbortError';
        throw e;
    }
}

function validateGraph(graph: StartupGraph): void {
    const nodes = graph.nodes;
    const ids = new Set(nodes.map((n) => n.id));
    if (ids.size !== nodes.length) {
        throw new Error('StartupGraph: 存在重复的节点 id');
    }
    for (const n of nodes) {
        if (!n.id) {
            throw new Error('StartupGraph: 节点缺少 id');
        }
        for (const d of n.dependsOn ?? []) {
            if (!ids.has(d)) {
                throw new Error(`StartupGraph: 节点 "${n.id}" 依赖不存在的节点 "${d}"`);
            }
        }
    }

    const nodeById = new Map(nodes.map((n) => [n.id, n]));
    const WHITE = 0;
    const GRAY = 1;
    const BLACK = 2;
    const color = new Map<string, number>();
    for (const n of nodes) {
        color.set(n.id, WHITE);
    }

    const dfs = (id: string): void => {
        const c = color.get(id) ?? WHITE;
        if (c === BLACK) {
            return;
        }
        if (c === GRAY) {
            throw new Error(`StartupGraph: 存在环，涉及节点 "${id}"`);
        }
        color.set(id, GRAY);
        const n = nodeById.get(id);
        for (const d of n?.dependsOn ?? []) {
            dfs(d);
        }
        color.set(id, BLACK);
    };

    for (const n of nodes) {
        dfs(n.id);
    }
}

/**
 * 按有向无环图执行：依赖全部完成后节点才开始；就绪节点之间并行。
 * 总进度 = Σ(weight × 节点进度) / Σ(weight)。
 */
export class StartupWorker {
    private state: StartupUIState = initialState();
    private readonly stateListeners = new Set<StartupStateListener>();
    private readonly overallListeners = new Set<StartupOverallListener>();
    private readonly enterGameListeners = new Set<StartupEnterGameListener>();

    constructor(private readonly graph: StartupGraph) {}

    getState(): Readonly<StartupUIState> {
        return this.state;
    }

    subscribeState(fn: StartupStateListener): () => void {
        this.stateListeners.add(fn);
        return () => this.stateListeners.delete(fn);
    }

    /** 仅总进度变化时回调，适合驱动加载条 */
    subscribeOverall(fn: StartupOverallListener): () => void {
        this.overallListeners.add(fn);
        return () => this.overallListeners.delete(fn);
    }

    onEnterGame(fn: StartupEnterGameListener): () => void {
        this.enterGameListeners.add(fn);
        return () => this.enterGameListeners.delete(fn);
    }

    private patch(partial: Partial<StartupUIState>): void {
        const prevOverall = this.state.overall;
        this.state = { ...this.state, ...partial };
        if ('overall' in partial && this.state.overall !== prevOverall) {
            for (const fn of this.overallListeners) {
                fn(this.state.overall);
            }
        }
        for (const fn of this.stateListeners) {
            fn(this.state);
        }
    }

    async run(options?: { signal?: AbortSignal }): Promise<void> {
        const signal = options?.signal;
        validateGraph(this.graph);

        this.state = initialState();
        this.patch({
            phase: 'running',
            overall: 0,
            error: null,
            activeId: null,
            activeLabel: null,
            activeProgress: 0,
            scopeId: null,
            scopeLabel: null,
            lastEvent: null,
        });

        const nodes = this.graph.nodes;
        const W = sumWeights(nodes);
        const nodeById = new Map(nodes.map((n) => [n.id, n]));
        const progress = new Map<string, number>();
        for (const n of nodes) {
            progress.set(n.id, 0);
        }

        const updateOverall = (): void => {
            if (W <= 0) {
                return;
            }
            let num = 0;
            for (const n of nodes) {
                num += (n.weight ?? 1) * (progress.get(n.id) ?? 0);
            }
            const o = clamp01(num / W);
            if (o !== this.state.overall) {
                this.patch({ overall: o });
            }
        };

        const completion = new Map<string, Promise<void>>();

        const runNode = (id: string): Promise<void> => {
            const cached = completion.get(id);
            if (cached) {
                return cached;
            }
            const node = nodeById.get(id);
            if (!node) {
                return Promise.resolve();
            }

            const p = (async () => {
                checkAborted(signal);
                const deps = node.dependsOn ?? [];
                await Promise.all(deps.map((depId) => runNode(depId)));
                checkAborted(signal);

                let last = 0;
                const ctx: StartupWorkContext = {
                    id: node.id,
                    label: node.label,
                    report: (pv: number) => {
                        last = clamp01(pv);
                        progress.set(node.id, last);
                        updateOverall();
                        this.patch({
                            activeId: node.id,
                            activeLabel: node.label ?? null,
                            activeProgress: last,
                        });
                    },
                    emit: (name: string, payload?: unknown) => {
                        this.patch({ lastEvent: { name, payload } });
                    },
                    signal,
                };

                this.patch({ lastEvent: { name: node.id, payload: 'start' } });

                try {
                    await node.work(ctx);
                } catch (err) {
                    const e = err instanceof Error ? err : new Error(String(err));
                    e.message = `[startup:${node.id}] ${e.message}`;
                    throw e;
                }

                if (last < 1) {
                    progress.set(node.id, 1);
                    updateOverall();
                    this.patch({
                        activeId: node.id,
                        activeLabel: node.label ?? null,
                        activeProgress: 1,
                    });
                }
            })();

            completion.set(id, p);
            return p;
        };

        try {
            checkAborted(signal);
            await Promise.all(nodes.map((n) => runNode(n.id)));

            this.patch({
                overall: 1,
                phase: 'success',
                activeId: null,
                activeLabel: null,
                activeProgress: 0,
                scopeId: null,
                scopeLabel: null,
            });

            const cbs = [...this.enterGameListeners];
            for (const fn of cbs) {
                fn();
            }
        } catch (e) {
            const err = e instanceof Error ? e : new Error(String(e));
            this.patch({
                phase: 'failed',
                error: err,
                activeId: null,
                activeProgress: 0,
            });
            throw e;
        }
    }
}