sync_replay_project/packages/server/lib/src/server_hub.dart

import 'dart:async';
import 'dart:convert';

import 'package:relic/relic.dart';
import 'package:sync_relay_core/core.dart';
import 'package:web_socket/web_socket.dart';

import 'relay_routes.dart';
import 'session.dart';

final _hubLog = createLogger('server_hub');

class ServerHub {
  final SessionMap sessions;
  final String expectedSecret;
  /// See [AppConfig.hostKey]; empty means nobody is the host.
  final String hostKey;
  final SyncEngine engine;
  final Map<String, RelicWebSocket> sockets = {};
  /// The server's Jellyfin login for the `/j/...` relay; null = no relay.
  JellyfinCredential? jellyfin;

  ServerHub(this.sessions, this.expectedSecret,
      {SyncEngine? engine, this.hostKey = '', this.jellyfin})
      : engine = engine ?? SyncEngine();

  /// The current media is a Jellyfin ref on the server [jellyfin] logs in to.
  bool get relayAvailable {
    final ref = session.media?.jellyfin;
    final cred = jellyfin;
    return ref != null && cred != null && cred.serverId.isNotEmpty && cred.serverId == ref.serverId;
  }

  StateBroadcast stateBroadcast() => engine.broadcast(relay: relayAvailable);

  /// The relay of the current media (`/r/...`), if it is relayed.
  ActiveRelay? relay;

  /// Replaces the relay (null: the new media is not relayed). The old
  /// buffer is dropped with it.
  void setRelay(ActiveRelay? next) {
    relay?.dispose();
    relay = next;
  }

  /// `media.relay` for a relayed swap by [token]: who serves the bytes.
  MediaRelay relayFor(String token, MediaRelay requested) => MediaRelay(
      source: requested.fetch ? '' : (engine.members[token]?.id ?? ''), fetch: requested.fetch);

  /// The relay's channel came up or went down: `media.relay.live` follows.
  void relayLiveChanged(ActiveRelay r) {
    final m = engine.media;
    if (!identical(relay, r) || m?.relay == null) return;
    if (m!.relay!.live == r.live) return;
    engine.media = m.copyWith(relay: m.relay!.copyWith(live: r.live));
    changed();
  }

  bool isHostKey(String key) => hostKey.isNotEmpty && key == hostKey;

  /// The member a reconnecting client may take over with its old token:
  /// still known and not kicked.
  EngineMember? resumable(String token) {
    if (token.isEmpty) return null;
    final viewer = session.tokens[token];
    if (viewer == null || viewer.kicked) return null;
    return engine.members[token];
  }

  Session get session => sessions.getOrCreate('default', expectedSecret);

  void broadcastState() {
    final raw = jsonEncode(stateBroadcast().toJson());
    for (final ws in sockets.values.toList()) {
      ws.trySendText(raw);
    }
  }

  void broadcastFeed() {
    for (final feed in engine.feed.drain()) {
      final raw = jsonEncode(feed.toJson());
      for (final ws in sockets.values.toList()) {
        ws.trySendText(raw);
      }
    }
  }

  void dropSilentSockets() {
    for (final token in engine.sweepDropped()) {
      final ws = sockets[token];
      if (ws != null) unawaited(ws.tryClose());
      forgetSocket(token);
    }
  }

  void sweepClosedSockets() {
    for (final entry in sockets.entries.toList()) {
      var closed = false;
      try {
        closed = entry.value.isClosed;
      } catch (_) {
        closed = true;
      }
      if (closed) forgetSocket(entry.key);
    }
  }

  Future<void> closeAllSockets() async {
    final all = sockets.values.toList();
    sockets.clear();
    for (final ws in all) {
      try {
        await ws.tryClose();
      } catch (_) {}
    }
  }

  void broadcastChat(String from, String text) {
    final clean = text.length > RelayConstants.maxChatLength
        ? text.substring(0, RelayConstants.maxChatLength)
        : text;
    final raw = jsonEncode(
        ChatMessage(text: clean).toJson()..addAll({'from': from}));
    for (final ws in sockets.values.toList()) {
      ws.trySendText(raw);
    }
    engine.feed.add(
        FeedMessage(actor: from, action: 'chat', at: engine.now));
    broadcastFeed();
  }

  void changed() {
    broadcastState();
    broadcastFeed();
  }

  /// Like [forgetSocket], but only while [ws] is still the member's socket:
  /// after a resume the old socket's close must not remove the member.
  void forgetSocketOf(String token, RelicWebSocket ws) {
    if (!identical(sockets[token], ws)) return;
    forgetSocket(token);
  }

  void forgetSocket(String token) {
    engine.applyLeave(token);
    final viewer = session.tokens[token];
    if (viewer != null && !viewer.kicked) session.tokens.remove(token);
    sockets.remove(token);
    changed();
  }

  Future<void> kickById(String targetId, String reason) async {
    final member = engine.byId(targetId);
    if (member == null) throw const AuthException('unknown kick target');
    if (session.tokens[member.token]?.isHost ?? false) {
      throw const AuthException("can't kick the host");
    }
    await session.kick(member.token, reason: reason);
  }
}

/// `GET /a/<sid>/<token>/<hash>`: a current member's picture by its
/// content hash, for members (valid, unkicked token) only. The bytes for a
/// hash never change, so clients may cache them for good.
void mountAvatarRoute(RelicApp app, ServerHub hub) {
  Response status(int code) => Response(code, body: Body.fromString('status $code'));
  app.get('/a/:sid/:token/:hash', (final req) {
    final sid = req.pathParameters.raw[#sid] ?? '';
    final token = req.pathParameters.raw[#token] ?? '';
    final hash = req.pathParameters.raw[#hash] ?? '';
    final session = hub.sessions.get(sid);
    if (session == null) return status(410);
    switch (session.gate(token)) {
      case MediaGate.unauthorized:
        return status(401);
      case MediaGate.forbidden:
        return status(403);
      default:
        break;
    }
    final bytes = hub.engine.avatarFor(hash);
    final type = bytes == null ? null : avatarImageType(bytes);
    if (bytes == null || type == null) return status(404);
    return Response.ok(
      body: Body.fromData(bytes, mimeType: MimeType(type.$1, type.$2)),
      headers: Headers.build((h) {
        h['cache-control'] = ['private, max-age=31536000, immutable'];
      }),
    );
  });
}

/// Media names are shown to everyone and must never carry a URL (api_key etc.).
/// A URL-looking name is reduced to its last path segment, no query/fragment.
String sanitizeMediaName(String name) {
  final trimmed = name.trim();
  if (!RegExp(r'^[a-zA-Z][a-zA-Z0-9+.-]*://').hasMatch(trimmed)) return name;
  final uri = Uri.tryParse(trimmed);
  final segments = uri?.pathSegments.where((s) => s.isNotEmpty).toList() ?? const [];
  if (segments.isNotEmpty) return segments.last;
  return 'media';
}

void wsErr(RelicWebSocket ws, String message) {
  ws.trySendText(jsonEncode({'t': 'err', 'message': message}));
}

/// Jellyfin ids are path segments of upstream requests: keep them plain.
final _jellyfinId = RegExp(r'^[A-Za-z0-9-]{1,64}$');

/// The swap's Jellyfin ref cleaned for broadcast (names like
/// [sanitizeMediaName]), or null when its ids are unusable.
JellyfinRef? sanitizeJellyfinRef(JellyfinRef ref) {
  if (!_jellyfinId.hasMatch(ref.serverId) || !_jellyfinId.hasMatch(ref.itemId)) return null;
  final source = ref.mediaSourceId;
  if (source != null && !_jellyfinId.hasMatch(source)) return null;
  final runtime = ref.runtimeSecs;
  return JellyfinRef(
    serverId: ref.serverId,
    itemId: ref.itemId,
    mediaSourceId: source,
    title: sanitizeMediaName(ref.title),
    subtitle: ref.subtitle == null ? null : sanitizeMediaName(ref.subtitle!),
    runtimeSecs: runtime == null || runtime < 0 ? null : runtime,
  );
}

/// [relayUrlBuilder] builds a member's `/j/...` URL for `welcome.relayUrl`.
void mountSyncRoute(
  RelicApp app,
  ServerHub hub,
  String Function(String sid, String token, SessionMedia? media, String host)
      urlBuilder, {
  String Function(String sid, String token, String itemId, String host)? relayUrlBuilder,
  String Function(String sid, String token, int mediaId, String name, String host)? relayMediaUrlBuilder,
}) {
  app.get('/sync', (final req) {
    // Host the client dialled (Host header), so file URLs work for remote joiners.
    final dialledHost = req.url.hasPort
        ? '${req.url.host.contains(':') ? '[${req.url.host}]' : req.url.host}:${req.url.port}'
        : (req.url.host.contains(':') ? '[${req.url.host}]' : req.url.host);
    return WebSocketUpgrade((final ws) async {
      ws.pingInterval = Duration(seconds: RelayConstants.pingIntervalSecs);
      final firstText = Completer<String>();
      String? token;
      var welcomed = false;
      late StreamSubscription<WebSocketEvent> sub;
      Future<void> close() async {
        try {
          await sub.cancel();
        } catch (_) {}
        await ws.tryClose();
      }

      void handleIntent(String text) {
        final session = hub.session;
        final viewer = token == null ? null : session.tokens[token];
        viewer?.touch();
        Map<String, Object?> decoded;
        try {
          final parsed = jsonDecode(text);
          if (parsed is! Map) return;
          decoded = Map<String, Object?>.from(parsed);
        } catch (_) {
          return;
        }
        final type = decoded['t'];
        try {
          switch (type) {
            case 'ready':
              if (token == null) return;
              final intent = ReadyIntent.fromJson(decoded);
              hub.engine.applyReady(token, intent.ready, source: intent.source);
              hub.changed();
            case 'force':
              if (token == null) return;
              final intent = ForceIntent.fromJson(decoded);
              hub.engine.applyForce(token, intent.target, intent.ready);
              hub.changed();
            case 'seek':
              if (token == null) return;
              final intent = SeekIntent.fromJson(decoded);
              hub.engine.applySeek(token, intent.position, seq: intent.seq);
              hub.changed();
            case 'swap':
              if (token == null) return;
              final parsed = SwapIntent.fromJson(decoded).media;
              if (parsed.jellyfin != null) {
                // By reference only: no URL (it could carry a token) is
                // stored or broadcast for Jellyfin media.
                final ref = sanitizeJellyfinRef(parsed.jellyfin!);
                // Relayed for viewers without access: from the swapper's
                // machine (its own sign-in never leaves it).
                final relayReq = parsed.relay == null || parsed.relay!.fetch ? null : parsed.relay;
                final media = ref == null
                    ? null
                    : HelloMedia(
                        name: sanitizeMediaName(parsed.name.isEmpty ? ref.title : parsed.name),
                        size: parsed.size,
                        duration: parsed.duration,
                        jellyfin: ref,
                        relay: relayReq == null ? null : hub.relayFor(token, relayReq));
                if (media == null || !hub.engine.applySwap(token, media)) {
                  wsErr(ws, 'bad swap media');
                  return;
                }
                session.media = SessionMedia(
                    name: media.name,
                    size: media.size,
                    duration: media.duration,
                    jellyfin: ref,
                    mediaId: hub.engine.mediaId);
                hub.setRelay(relayReq == null ? null : _memberRelay(hub, token, media.name));
                hub.changed();
                return;
              }
              final relayReq = parsed.relay;
              if (relayReq != null) {
                // Relayed file or link: no URL is stored or broadcast (a
                // link may be private to the swapper; the server keeps a
                // fetched one to itself).
                final name = sanitizeMediaName(parsed.name.isEmpty ? 'media' : parsed.name);
                final link = Uri.tryParse(parsed.url ?? '');
                if (relayReq.fetch &&
                    (link == null || !(link.isScheme('http') || link.isScheme('https')) || link.host.isEmpty)) {
                  wsErr(ws, 'relay fetch needs an http(s) url');
                  return;
                }
                // The server would fetch any address it can reach (its own
                // network, a VPN) for whoever asks: only for members who
                // know the password.
                if (relayReq.fetch && session.secret.isEmpty) {
                  wsErr(ws, 'server fetch needs a server password');
                  return;
                }
                final media = HelloMedia(
                    name: name,
                    size: parsed.size,
                    duration: parsed.duration,
                    relay: hub.relayFor(token, relayReq));
                if (!hub.engine.applySwap(token, media)) {
                  wsErr(ws, 'bad swap media');
                  return;
                }
                session.media = SessionMedia(
                    name: name, size: parsed.size, duration: parsed.duration, mediaId: hub.engine.mediaId);
                late final ActiveRelay active;
                active = relayReq.fetch
                    ? ActiveRelay.fetch(hub.engine.mediaId, name, link!,
                        onLiveChanged: () => hub.relayLiveChanged(active))
                    : _memberRelay(hub, token, name);
                hub.setRelay(active);
                hub.changed();
                return;
              }
              final intent = SwapIntent(
                  media: HelloMedia(
                      name: sanitizeMediaName(parsed.name),
                      size: parsed.size,
                      duration: parsed.duration,
                      url: parsed.url));
              final current = session.media;
              final matchesFile = current?.file != null &&
                  current!.name == intent.media.name &&
                  current.size == intent.media.size;
              if (!matchesFile && (intent.media.url ?? '').isEmpty) {
                wsErr(ws, 'swap needs a url or the hosted file');
                return;
              }
              if (!hub.engine.applySwap(token, intent.media)) {
                wsErr(ws, 'bad swap media');
                return;
              }
              hub.setRelay(null);
              if (matchesFile) {
                session.media = SessionMedia(
                    name: current.name,
                    size: current.size,
                    duration: intent.media.duration,
                    url: intent.media.url,
                    file: current.file,
                    mediaId: hub.engine.mediaId);
              } else {
                session.media = SessionMedia(
                    name: intent.media.name,
                    size: intent.media.size,
                    duration: intent.media.duration,
                    url: intent.media.url,
                    mediaId: hub.engine.mediaId);
              }
              hub.changed();
            case 'profile':
              if (token == null || viewer == null) return;
              final intent = ProfileIntent.fromJson(decoded);
              final avatar = intent.avatar;
              if (avatar != null && avatar.length > RelayConstants.maxAvatarBytes) {
                wsErr(ws, 'rejected: avatar too large');
                return;
              }
              final name = intent.name == null ? null : cleanMemberName(intent.name!);
              // Chat and kick name the sender by its token's name.
              if (name != null) viewer.name = name;
              hub.engine.applyProfile(token, name: name, avatar: avatar);
              hub.changed();
            case 'report':
              if (token == null) return;
              final report = ReportMessage.fromJson(decoded);
              hub.engine.applyReport(
                  token, report.position, report.buffered, report.eventId,
                  duration: report.duration, mediaId: report.mediaId, jf: report.jf);
              hub.changed();
            case 'ping':
              final ping = PingMessage.fromJson(decoded);
              ws.trySendText(jsonEncode(PongMessage(
                      clientMs: ping.clientMs,
                      serverMs: hub.engine.now.millisecondsSinceEpoch)
                  .toJson()));
            case 'chat':
              final chat = ChatMessage.fromJson(decoded);
              hub.broadcastChat(viewer?.name ?? '?', chat.text);
            case 'kick':
              final kickReq = KickRequest.fromJson(decoded);
              final member = hub.engine.byId(kickReq.target);
              if (member == null) {
                wsErr(ws, 'unknown kick target');
                return;
              }
              if (session.tokens[member.token]?.isHost ?? false) {
                wsErr(ws, "can't kick the host");
                return;
              }
              final by = viewer?.name ?? '?';
              unawaited(hub.session.kick(member.token,
                  reason: kickReq.reason.isEmpty
                      ? 'kicked by $by'
                      : kickReq.reason));
          }
        } on ProtocolException catch (e) {
          wsErr(ws, 'bad message: ${e.message}');
        } catch (_) {
          return;
        }
      }

      sub = ws.events.listen(
        (event) {
          if (event is! TextDataReceived) return;
          if (!firstText.isCompleted) {
            firstText.complete(event.text);
          } else if (welcomed) {
            handleIntent(event.text);
          }
        },
        onError: (_) {
          if (token != null) hub.forgetSocketOf(token, ws);
        },
        onDone: () {
          if (!firstText.isCompleted) {
            firstText.completeError(StateError('closed before hello'));
          } else if (token != null) {
            hub.forgetSocketOf(token, ws);
          }
        },
      );
      String raw;
      try {
        raw = await firstText.future.timeout(const Duration(seconds: 10));
      } catch (_) {
        await close();
        return;
      }
      HelloMessage hello;
      try {
        final decoded = jsonDecode(raw);
        if (decoded is! Map) throw const ProtocolException('hello must be an object');
        hello = HelloMessage.fromJson(Map<String, Object?>.from(decoded));
      } catch (e) {
        _hubLog.warning('reject: bad hello: $e');
        wsErr(ws, 'bad hello: $e');
        await close();
        return;
      }
      if (hello.proto != RelayConstants.protoVersion) {
        _hubLog.warning(
            'reject ${hello.user.name}: proto mismatch: got ${hello.proto}, want ${RelayConstants.protoVersion}');
        wsErr(ws, 'proto mismatch: got ${hello.proto}, want ${RelayConstants.protoVersion}');
        await close();
        return;
      }
      final session = hub.session;
      if (session.secret.isNotEmpty && hello.secret != session.secret) {
        _hubLog.warning('reject ${hello.user.name}: wrong secret');
        wsErr(ws, 'wrong secret');
        await close();
        return;
      }
      final resumed = hub.resumable(hello.resume);
      String memberId;
      if (resumed != null) {
        // Unclean drop the server has not noticed yet: take the member over
        // (same token, id, ready flag) and drop the dead socket quietly.
        token = hello.resume;
        memberId = resumed.id;
        final old = hub.sockets.remove(token);
        if (old != null) unawaited(old.tryClose());
        hub.engine.applyResume(token);
        _hubLog.info('resume ${hello.user.name} ($memberId)');
      } else {
        try {
          token = session.issueToken(hello.user.name, hello.user.avatar);
        } catch (e) {
          _hubLog.warning('reject ${hello.user.name}: $e');
          wsErr(ws, 'rejected: $e');
          await close();
          return;
        }
        memberId = hub.engine.applyJoin(token, session.tokens[token]!.name, hello.user.avatar);
      }
      session.tokens[token]!.isHost = hub.isHostKey(hello.hostKey);
      final activeToken = token;
      hub.sockets[activeToken] = ws;
      session.tokens[activeToken]!.onKick = (reason) async {
        if (hub.relay?.sourceToken == activeToken) hub.relay!.dropChannel();
        ws.trySendText(jsonEncode(KickedMessage(reason: reason).toJson()));
        await Future.delayed(const Duration(milliseconds: 200));
        await ws.tryClose();
        hub.forgetSocketOf(activeToken, ws);
      };
      final media = session.media;
      final relay = hub.relayAvailable;
      final relayed = hub.relay;
      final welcome = WelcomeMessage(
        token: activeToken,
        memberId: memberId,
        // Jellyfin media resolves per member (own account first); only
        // relayed files and links start from the /r/ URL, and never for
        // the source itself (it plays its own copy).
        mediaUrl: relayed != null &&
                media != null &&
                media.jellyfin == null &&
                relayed.sourceToken != activeToken &&
                relayed.mediaId == media.mediaId &&
                relayMediaUrlBuilder != null
            ? relayMediaUrlBuilder('default', activeToken, relayed.mediaId, relayed.name, dialledHost)
            : media?.file != null
                ? urlBuilder('default', activeToken, media, dialledHost)
                : media?.url,
        relayUrl: relay && relayUrlBuilder != null
            ? relayUrlBuilder('default', activeToken, media!.jellyfin!.itemId, dialledHost)
            : null,
        state: hub.stateBroadcast().toJson(),
      );
      try {
        ws.sendText(jsonEncode(welcome.toJson()));
      } catch (_) {
        await close();
        return;
      }
      welcomed = true;
      _hubLog.info('join ${hello.user.name} ($memberId)');
      hub.changed();
      try {
        await sub.asFuture();
      } catch (_) {}
    });
  });
}

ActiveRelay _memberRelay(ServerHub hub, String token, String name) {
  late final ActiveRelay r;
  r = ActiveRelay.fromMember(hub.engine.mediaId, name, token, onLiveChanged: () => hub.relayLiveChanged(r));
  return r;
}