SoulStream/src/lib/preview/service.js

// Preview session service — the effectful orchestrator over the pure parts.
//
// The reducer decides WHAT happens; this service decides WHEN: the settle
// window after navigation, the resolve→load→play pipeline, the single
// cache-bypass retry, one direction-aware lease prefetch, and the focus
// handshake at the session edges. Everything time-like is injected
// (scheduler, monotonic clock) so the whole loop runs under node:test with
// manual ticks — no real timers anywhere in the tests.
//
// Staleness discipline: every async continuation re-checks BOTH the session
// id and the request generation before dispatching or touching the
// transport. The reducer would drop a stale dispatch anyway; checking here
// too keeps the transport from even being commanded on behalf of a dead
// generation (the "transport boundary" half of the Phase 2 exit gate).

import { reduce, createInitialState, playableManifestIndices, playableCursor, PHASE } from "./session.js";

const defaultScheduler = {
    set: (fn, ms) => setTimeout(fn, ms),
    clear: (id) => clearTimeout(id),
};

export function createPreviewService({
    resolver,
    // The service CONSTRUCTS the transport, binding it to its own dispatch —
    // the transport's events (MEDIA_PLAYING, MEDIA_ENDED, …) flow through the
    // same generation-guarded reducer as everything else. Passing an already-
    // built transport is how the original draft silently lost that wiring.
    transportFactory,
    focus, // { coordinator } from focusBridge, or a test double
    settleMs = 150,
    scheduler = defaultScheduler,
    monotonicNow = () => (globalThis.performance ?? Date).now(),
}) {
    let state = null;
    let listeners = new Set();
    let settleTimer = null;
    let focusToken = null;
    let focusWaiting = false;
    // sessionKey -> { lease, receiptAt } — the one direction-aware prefetch.
    let prefetched = new Map();

    const emit = () => {
        for (const l of listeners) l(state, { focusWaiting });
    };
    const dispatch = (action) => {
        if (!state) return;
        const before = state;
        state = reduce(state, action);
        if (state !== before) {
            afterChange(before);
            emit();
        }
    };
    const transport = transportFactory(dispatch);

    // ------------------------------------------------------ media pipeline --

    function afterChange(before) {
        if (!state.manifest) return;
        const generationMoved = state.requestGeneration !== before.requestGeneration;
        if (generationMoved && state.phase === PHASE.RESOLVING_MEDIA) {
            scheduleSettle();
        }
    }

    function scheduleSettle() {
        if (settleTimer != null) scheduler.clear(settleTimer);
        const sessionId = state.sessionId;
        const generation = state.requestGeneration;
        // Stop the outgoing clip immediately — silence between cards is
        // correct; audio under the wrong artwork never is.
        void transport.stop();
        settleTimer = scheduler.set(() => {
            settleTimer = null;
            if (!state || state.sessionId !== sessionId || state.requestGeneration !== generation) return;
            void runPipeline(sessionId, generation);
        }, settleMs);
    }

    async function runPipeline(sessionId, generation) {
        const fresh = () =>
            state && state.sessionId === sessionId && state.requestGeneration === generation;
        if (!fresh() || state.activeItemKey == null) return;
        const item = state.manifest.items[state.activeManifestIndex];
        const source = item.sources[0];
        if (!source) {
            dispatch({ type: "ITEM_FAILED", sessionId, generation, sessionKey: item.sessionKey, errorClass: "no_source" });
            return;
        }
        dispatch({ type: "RESOLVE_STARTED", sessionId, generation, sessionKey: item.sessionKey });

        // Prefetched lease still inside its safe window → skip the resolve.
        let lease = null;
        const pre = prefetched.get(item.sessionKey);
        if (pre && monotonicNow() - pre.receiptAt < pre.lease.safeForMs * 0.8) {
            lease = pre.lease;
        }
        prefetched.delete(item.sessionKey);

        if (!lease) {
            let result = await resolver.resolve(source, { generation });
            if (!fresh()) return;
            if (!result.ok && result.retryable) {
                // The capped retry: exactly one cache-bypass re-mint per
                // user intent, per the resolver contract. Never a loop.
                result = await resolver.resolve(source, { generation, bypassCache: true });
                if (!fresh()) return;
            }
            if (!result.ok) {
                dispatch({
                    type: "ITEM_FAILED",
                    sessionId,
                    generation,
                    sessionKey: item.sessionKey,
                    errorClass: result.failure,
                    availability: result.failure === "not_provided" ? "not_provided" : undefined,
                });
                return;
            }
            lease = result.lease;
        }

        dispatch({
            type: "LEASE_READY",
            sessionId,
            generation,
            sessionKey: item.sessionKey,
            clipDurationMs: lease.clipDurationMs,
        });
        if (!fresh()) return;

        const audible = state.desiredPlaying && focusToken != null && !state.muted;
        await transport.load(lease, { sessionId, generation, sessionKey: item.sessionKey }, { startPaused: !audible });
        if (!fresh()) {
            void transport.stop();
            return;
        }
        if (audible) await transport.play();

        prefetchNeighbor(sessionId, generation);
    }

    /** Resolve the NEXT playable item's lease in the background so stepping
     * forward usually skips the network entirely. One item, forward only —
     * the bounded mechanism the plan allows without byte prefetch. */
    function prefetchNeighbor(sessionId, generation) {
        if (!state || state.sessionId !== sessionId || state.requestGeneration !== generation) return;
        const playable = playableManifestIndices(state);
        const cur = playableCursor(state);
        if (cur == null || cur + 1 >= playable.length) return;
        const next = state.manifest.items[playable[cur + 1]];
        if (prefetched.has(next.sessionKey) || !next.sources[0]) return;
        void resolver
            .resolve(next.sources[0], { generation })
            .then((r) => {
                if (r.ok && state && state.sessionId === sessionId) {
                    prefetched.set(next.sessionKey, { lease: r.lease, receiptAt: monotonicNow() });
                    // bounded: keep at most one prefetched lease around
                    if (prefetched.size > 1) {
                        const oldest = prefetched.keys().next().value;
                        if (oldest !== next.sessionKey) prefetched.delete(oldest);
                    }
                }
            })
            .catch(() => {});
    }

    // -------------------------------------------------------------- focus --

    async function ensureFocus() {
        if (focusToken) return true;
        focusWaiting = true;
        emit();
        try {
            const { granted, token } = await focus.coordinator.acquirePreview();
            focusToken = granted ? token : null;
            return granted;
        } catch {
            focusToken = null;
            return false;
        } finally {
            focusWaiting = false;
            emit();
        }
    }

    // ----------------------------------------------------------- public ----

    return {
        subscribe(fn) {
            listeners.add(fn);
            if (state) fn(state, { focusWaiting });
            return () => listeners.delete(fn);
        },
        getState: () => state,
        isFocusWaiting: () => focusWaiting,

        /** Open a session: focus first (take-audio-on-open), then manifest.
         * If the acknowledgement fails, the session still opens — parked,
         * with the transient wait surfaced — and Play retries acquisition. */
        async open({ sessionId, manifest, startAtManifestIndex }) {
            state = createInitialState(sessionId);
            emit();
            const granted = await ensureFocus();
            dispatch({ type: "MANIFEST_READY", sessionId, manifest, startAtManifestIndex });
            if (!granted && state.phase !== PHASE.EMPTY) {
                dispatch({ type: "SET_DESIRED_PLAYING", sessionId, value: false });
            }
        },

        step: (dir) => dispatch({ type: "STEP", sessionId: state?.sessionId, dir }),
        activateKey: (sessionKey) => dispatch({ type: "ACTIVATE_KEY", sessionId: state?.sessionId, sessionKey }),
        replay: () => dispatch({ type: "REPLAY", sessionId: state?.sessionId }),
        seek: (positionMs) => {
            if (!state) return;
            void transport.seek?.(positionMs);
            dispatch({
                type: "MEDIA_POSITION",
                sessionId: state.sessionId,
                generation: state.requestGeneration,
                positionMs,
            });
        },
        setMuted(value) {
            if (!state) return;
            dispatch({ type: "SET_MUTED", sessionId: state.sessionId, value });
            if (value && state.phase === PHASE.PLAYING) void transport.pause();
        },

        async togglePlay() {
            if (!state) return;
            const sid = state.sessionId;
            if (state.phase === PHASE.PLAYING) {
                dispatch({ type: "SET_DESIRED_PLAYING", sessionId: sid, value: false });
                await transport.pause();
                return;
            }
            if (state.phase === PHASE.ENDED) {
                dispatch({ type: "REPLAY", sessionId: sid });
                return;
            }
            dispatch({ type: "SET_DESIRED_PLAYING", sessionId: sid, value: true });
            const granted = await ensureFocus();
            if (!granted) return; // parked; wait state already surfaced
            if (state.phase === PHASE.READY || state.phase === PHASE.PAUSED) {
                await transport.play();
            }
            // RESOLVING/BUFFERING: the pipeline observes desiredPlaying when
            // the lease lands; nothing else to do here.
        },

        /** Play full track: hand ownership to ordinary playback. The old
         * main snapshot is intentionally discarded (never restored). */
        async playFullTrack(startMainPlayback) {
            if (!state) return;
            await transport.stop();
            focus.coordinator.handoffToMain();
            focusToken = null;
            dispatch({ type: "SET_DESIRED_PLAYING", sessionId: state.sessionId, value: false });
            await startMainPlayback(state.manifest.items[state.activeManifestIndex]);
        },

        /** Close: terminal for the session; transport torn down; main
         * restored only if we paused it and nothing changed since. */
        async close() {
            if (!state) return { restored: false };
            if (settleTimer != null) {
                scheduler.clear(settleTimer);
                settleTimer = null;
            }
            dispatch({ type: "CLOSE", sessionId: state.sessionId });
            prefetched.clear();
            await transport.stop();
            await transport.dispose();
            const token = focusToken;
            focusToken = null;
            const restored = token ? focus.coordinator.release(token) : { restored: false };
            return restored;
        },
    };
}