♻️ refactor(realtime): 拆分会话编排职责

backend/src/main/java/com/babelflux/backend/websocket/HandoffSessionSubscriber.java

@@ -0,0 +1,69 @@
+package com.babelflux.backend.websocket;
+
+import jakarta.annotation.PreDestroy;
+import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.springframework.web.socket.WebSocketSession;
+
+/** Owns read-only handoff subscriptions and their lifecycle. */
+final class HandoffSessionSubscriber {
+    private final SessionEventHub eventHub;
+    private final WebSocketEventSender sender;
+    private final ConcurrentMap<String, SessionEventHub.Subscription> subscriptions = new ConcurrentHashMap<>();
+    private final ConcurrentMap<String, Future<?>> tasks = new ConcurrentHashMap<>();
+    private final ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();
+
+    HandoffSessionSubscriber(SessionEventHub eventHub, WebSocketEventSender sender) {
+        this.eventHub = eventHub;
+        this.sender = sender;
+    }
+
+    boolean isHandoff(String socketId) { return subscriptions.containsKey(socketId); }
+
+    void start(WebSocketSession socket, String sessionId) throws IOException {
+        SessionEventHub.Subscription subscription = eventHub.subscribe(sessionId);
+        subscriptions.put(socket.getId(), subscription);
+        try {
+            sender.send(socket, Map.of("type", "session_started", "sessionId", sessionId));
+            for (Map<String, Object> event : subscription.replay()) sender.send(socket, event);
+            tasks.put(socket.getId(), executor.submit(() -> forward(socket, subscription)));
+        } catch (IOException error) {
+            close(socket.getId());
+            throw error;
+        }
+    }
+
+    boolean close(String socketId) {
+        SessionEventHub.Subscription subscription = subscriptions.remove(socketId);
+        if (subscription == null) return false;
+        eventHub.unsubscribe(subscription);
+        Future<?> task = tasks.remove(socketId);
+        if (task != null) task.cancel(true);
+        return true;
+    }
+
+    @PreDestroy
+    void shutdown() {
+        tasks.values().forEach(task -> task.cancel(true));
+        executor.shutdownNow();
+    }
+
+    private void forward(WebSocketSession socket, SessionEventHub.Subscription subscription) {
+        try {
+            while (!Thread.currentThread().isInterrupted()) {
+                Map<String, Object> event = subscription.await(1, TimeUnit.SECONDS);
+                if (event != null) sender.send(socket, event);
+            }
+        } catch (InterruptedException error) {
+            Thread.currentThread().interrupt();
+        } catch (Exception ignored) {
+            // A display-only socket can close while its primary session continues.
+        }
+    }
+}

backend/src/main/java/com/babelflux/backend/websocket/RealtimeSessionCoordinator.java

@@ -0,0 +1,175 @@
+package com.babelflux.backend.websocket;
+
+import com.babelflux.backend.domain.Session;
+import com.babelflux.backend.domain.SessionReport;
+import com.babelflux.backend.service.RealtimeSessionRunner;
+import com.babelflux.backend.service.SessionService;
+import jakarta.annotation.PreDestroy;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+/** Application coordinator for one realtime session runner; WebSocket parsing stays in the adapter. */
+final class RealtimeSessionCoordinator {
+    @FunctionalInterface
+    interface EventSink { void emit(Map<String, Object> event) throws Exception; }
+
+    record StartRequest(String sourceLanguage, String targetLanguage, String domain, String inputMode,
+                        String sourceUrl, String modelProfile) {}
+    record CommandResult(String error, SessionReport report) {
+        static CommandResult ok() { return new CommandResult(null, null); }
+        static CommandResult report(SessionReport report) { return new CommandResult(null, report); }
+        static CommandResult error(String error) { return new CommandResult(error, null); }
+        boolean failed() { return error != null; }
+    }
+
+    private final SessionService sessions;
+    private final RealtimeSessionRunner runner;
+    private final RunnerLeaseManager leases;
+    private final ConcurrentMap<String, RealtimeSessionRunner.RunHandle> runs = new ConcurrentHashMap<>();
+    private final ConcurrentMap<String, EventSink> eventSinks = new ConcurrentHashMap<>();
+
+    RealtimeSessionCoordinator(SessionService sessions, RealtimeSessionRunner runner, RunnerLeaseManager leases) {
+        this.sessions = sessions;
+        this.runner = runner;
+        this.leases = leases;
+    }
+
+    CommandResult start(String sessionId, String socketId, StartRequest request, EventSink sink) {
+        if (!SessionService.isSupportedInputMode(request.inputMode())) {
+            return CommandResult.error("暂不支持的输入模式：" + request.inputMode());
+        }
+        synchronized (runs) {
+            if (runs.containsKey(socketId)) return CommandResult.ok();
+            RunnerLeaseManager.AcquireResult acquired = leases.acquire(sessionId, socketId);
+            if (acquired != RunnerLeaseManager.AcquireResult.ACQUIRED) {
+                return CommandResult.error(acquireMessage(acquired));
+            }
+            Session session;
+            try {
+                session = sessions.get(sessionId);
+            } catch (RuntimeException | Error error) {
+                leases.release(sessionId, socketId);
+                throw error;
+            }
+            session.applyOverrides(request.sourceLanguage(), request.targetLanguage(), request.domain(),
+                    request.inputMode(), request.sourceUrl(), request.modelProfile());
+            session.start();
+            AtomicBoolean reportEmitted = new AtomicBoolean();
+            AtomicReference<RealtimeSessionRunner.RunHandle> handleRef = new AtomicReference<>();
+            try {
+                // Save the lifecycle fact before another instance can read stale session history.
+                sessions.saveProgress(session);
+                eventSinks.put(socketId, sink);
+                RealtimeSessionRunner.RunHandle handle = runner.start(session, event -> {
+                    if ("session_report".equals(event.get("type"))) {
+                        reportEmitted.set(true);
+                        RealtimeSessionRunner.RunHandle current = handleRef.get();
+                        if (current != null) runs.remove(socketId, current);
+                        leases.release(sessionId, socketId);
+                        eventSinks.remove(socketId, sink);
+                    }
+                    sink.emit(event);
+                });
+                handleRef.set(handle);
+                runs.put(socketId, handle);
+                if (reportEmitted.get()) {
+                    runs.remove(socketId, handle);
+                    leases.release(sessionId, socketId);
+                    eventSinks.remove(socketId, sink);
+                } else {
+                    leases.scheduleRenewal(sessionId, socketId, () -> leaseLost(sessionId, socketId));
+                }
+                return CommandResult.ok();
+            } catch (RuntimeException | Error error) {
+                eventSinks.remove(socketId, sink);
+                session.end();
+                try {
+                    sessions.saveProgress(session);
+                } catch (RuntimeException | Error ignored) {
+                    error.addSuppressed(ignored);
+                }
+                leases.release(sessionId, socketId);
+                throw error;
+            }
+        }
+    }
+
+    CommandResult stop(String sessionId, String socketId) {
+        RealtimeSessionRunner.RunHandle handle = runs.get(socketId);
+        if (handle != null) {
+            handle.stop();
+            return CommandResult.ok();
+        }
+        RunnerLeaseManager.FinishPermission permission = leases.canFinish(sessionId, socketId);
+        if (permission != RunnerLeaseManager.FinishPermission.ALLOWED) {
+            return CommandResult.error(finishMessage(permission));
+        }
+        return CommandResult.report(sessions.finish(sessionId));
+    }
+
+    void acceptAudio(String socketId, byte[] pcm) {
+        RealtimeSessionRunner.RunHandle handle = runs.get(socketId);
+        if (handle != null) handle.acceptAudio(pcm);
+    }
+
+    boolean hasRun(String socketId) { return runs.containsKey(socketId); }
+
+    void updateClock(String socketId, long playbackMs, long sentAudioMs) {
+        RealtimeSessionRunner.RunHandle handle = runs.get(socketId);
+        if (handle != null) handle.updateClientClock(playbackMs, sentAudioMs);
+    }
+
+    void pause(String socketId) {
+        RealtimeSessionRunner.RunHandle handle = runs.get(socketId);
+        if (handle != null) handle.pause();
+    }
+
+    void resume(String socketId) {
+        RealtimeSessionRunner.RunHandle handle = runs.get(socketId);
+        if (handle != null) handle.resume();
+    }
+
+    void close(String sessionId, String socketId) {
+        eventSinks.remove(socketId);
+        RealtimeSessionRunner.RunHandle handle = runs.remove(socketId);
+        if (handle != null) handle.stop();
+        leases.release(sessionId, socketId);
+    }
+
+    @PreDestroy
+    void shutdown() { eventSinks.clear(); }
+
+    private void leaseLost(String sessionId, String socketId) {
+        RealtimeSessionRunner.RunHandle handle = runs.get(socketId);
+        if (handle != null) handle.stop();
+        leases.release(sessionId, socketId);
+        EventSink sink = eventSinks.remove(socketId);
+        if (sink == null) return;
+        try {
+            sink.emit(Map.of("type", "error", "message", "会话租约续期失败，已停止实时会话"));
+        } catch (Exception ignored) {
+            // Socket ownership has already been released; sending the notice is best effort.
+        }
+    }
+
+    private static String acquireMessage(RunnerLeaseManager.AcquireResult result) {
+        return switch (result) {
+            case LOCAL_BUSY -> "会话已在其他连接中运行";
+            case REMOTE_BUSY -> "会话已在其他实例中运行";
+            case UNAVAILABLE -> "会话锁服务暂不可用";
+            case ACQUIRED -> throw new IllegalArgumentException("acquired has no failure message");
+        };
+    }
+
+    private static String finishMessage(RunnerLeaseManager.FinishPermission permission) {
+        return switch (permission) {
+            case LOCAL_BUSY -> "会话正在其他连接中运行";
+            case REMOTE_BUSY -> "会话正在其他实例中运行";
+            case UNAVAILABLE -> "会话锁服务暂不可用";
+            case ALLOWED -> throw new IllegalArgumentException("allowed has no failure message");
+        };
+    }
+}

backend/src/main/java/com/babelflux/backend/websocket/RunnerLeaseManager.java

@@ -0,0 +1,91 @@
+package com.babelflux.backend.websocket;
+
+import com.babelflux.backend.infrastructure.RedisSessionRepository;
+import jakarta.annotation.PreDestroy;
+import java.time.Duration;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+
+/** Coordinates local ownership and the optional Redis runner lease without touching WebSocket protocol code. */
+final class RunnerLeaseManager {
+    private static final long TTL_SECONDS = 30;
+    private static final long RENEW_SECONDS = 10;
+
+    enum AcquireResult { ACQUIRED, LOCAL_BUSY, REMOTE_BUSY, UNAVAILABLE }
+    enum FinishPermission { ALLOWED, LOCAL_BUSY, REMOTE_BUSY, UNAVAILABLE }
+
+    private final RedisSessionRepository redis;
+    private final ConcurrentMap<String, String> owners = new ConcurrentHashMap<>();
+    private final ConcurrentMap<String, ScheduledFuture<?>> renewals = new ConcurrentHashMap<>();
+    private final ScheduledExecutorService executor;
+
+    RunnerLeaseManager(RedisSessionRepository redis) {
+        this.redis = redis;
+        this.executor = redis == null ? null
+                : Executors.newSingleThreadScheduledExecutor(Thread.ofVirtual().name("runner-lease-renew").factory());
+    }
+
+    AcquireResult acquire(String sessionId, String socketId) {
+        if (owners.putIfAbsent(sessionId, socketId) != null) return AcquireResult.LOCAL_BUSY;
+        if (redis == null) return AcquireResult.ACQUIRED;
+        try {
+            if (redis.tryAcquireRunnerLease(sessionId, socketId, Duration.ofSeconds(TTL_SECONDS))) {
+                return AcquireResult.ACQUIRED;
+            }
+            owners.remove(sessionId, socketId);
+            return AcquireResult.REMOTE_BUSY;
+        } catch (RuntimeException error) {
+            owners.remove(sessionId, socketId);
+            return AcquireResult.UNAVAILABLE;
+        }
+    }
+
+    FinishPermission canFinish(String sessionId, String socketId) {
+        String owner = owners.get(sessionId);
+        if (owner != null && !owner.equals(socketId)) return FinishPermission.LOCAL_BUSY;
+        if (owner != null || redis == null) return FinishPermission.ALLOWED;
+        try {
+            return redis.runnerLeaseHeld(sessionId) ? FinishPermission.REMOTE_BUSY : FinishPermission.ALLOWED;
+        } catch (RuntimeException error) {
+            return FinishPermission.UNAVAILABLE;
+        }
+    }
+
+    void scheduleRenewal(String sessionId, String socketId, Runnable onRenewalLost) {
+        if (redis == null || executor == null) return;
+        ScheduledFuture<?> task = executor.scheduleAtFixedRate(() -> {
+            if (!socketId.equals(owners.get(sessionId))) return;
+            boolean renewed;
+            try {
+                renewed = redis.renewRunnerLease(sessionId, socketId, Duration.ofSeconds(TTL_SECONDS));
+            } catch (RuntimeException error) {
+                renewed = false;
+            }
+            if (!renewed) onRenewalLost.run();
+        }, RENEW_SECONDS, RENEW_SECONDS, TimeUnit.SECONDS);
+        ScheduledFuture<?> previous = renewals.put(sessionId, task);
+        if (previous != null) previous.cancel(false);
+    }
+
+    void release(String sessionId, String socketId) {
+        if (!owners.remove(sessionId, socketId)) return;
+        ScheduledFuture<?> renewal = renewals.remove(sessionId);
+        if (renewal != null) renewal.cancel(false);
+        if (redis == null) return;
+        try {
+            redis.releaseRunnerLease(sessionId, socketId);
+        } catch (RuntimeException ignored) {
+            // Redis TTL expires a lease that cannot be released during cleanup.
+        }
+    }
+
+    @PreDestroy
+    void shutdown() {
+        renewals.values().forEach(task -> task.cancel(true));
+        if (executor != null) executor.shutdownNow();
+    }
+}

backend/src/main/java/com/babelflux/backend/websocket/SessionEventDispatcher.java

@@ -0,0 +1,32 @@
+package com.babelflux.backend.websocket;
+
+import java.io.IOException;
+import java.util.Map;
+import org.springframework.web.socket.WebSocketSession;
+
+/** Publishes session events for handoff subscribers before sending to the primary socket. */
+final class SessionEventDispatcher {
+    private final SessionEventHub eventHub;
+    private final WebSocketEventSender sender;
+
+    SessionEventDispatcher(SessionEventHub eventHub, WebSocketEventSender sender) {
+        this.eventHub = eventHub;
+        this.sender = sender;
+    }
+
+    void sendDirect(WebSocketSession socket, Object body) throws IOException { sender.send(socket, body); }
+
+    void dispatchRunnerEvent(String sessionId, WebSocketSession socket, Map<String, Object> event) throws IOException {
+        eventHub.publish(sessionId, event);
+        if ("session_report".equals(event.get("type"))) eventHub.complete(sessionId);
+        sender.send(socket, event);
+    }
+
+    void dispatchSourceState(String sessionId, WebSocketSession socket, String status, String message)
+            throws IOException {
+        Map<String, Object> event = Map.of("type", "source_sync_state",
+                "state", Map.of("status", status, "lagMs", 0, "message", message));
+        eventHub.publish(sessionId, event);
+        sender.send(socket, event);
+    }
+}