fix: Stream build logs over WS

packages/globe_cli/lib/src/commands/build_logs_command.dart

@@ -1,6 +1,3 @@
-import 'dart:io';
-
-import 'package:eventsource/eventsource.dart';
 import 'package:mason_logger/mason_logger.dart';
 
 import '../command.dart';
@@ -25,10 +22,6 @@ class BuildLogsCommand extends BaseGlobeCommand {
   Future<int> run() async {
     requireAuth();
 
-    if (!scope.hasScope()) {
-      logger.err('Not a Globe project.');
-    }
-
     final validated = await scope.validate();
     final deploymentId = argResults!['deployment'] as String?;
 
@@ -39,24 +32,25 @@ class BuildLogsCommand extends BaseGlobeCommand {
       return ExitCode.software.code;
     }
 
-    late Stream<BuildLogEvent> logs;
+    final deployment = await api.getDeployment(
+      orgId: validated.organization.id,
+      projectId: validated.project.id,
+      deploymentId: deploymentId,
+    );
 
-    try {
-      logs = await streamBuildLogs(
-        api: api,
-        orgId: validated.organization.id,
-        projectId: validated.project.id,
-        deploymentId: deploymentId,
-      );
-    } on EventSourceSubscriptionException catch (e) {
-      logger.err(
-        e.statusCode == HttpStatus.notFound
-            ? '✗ An error occurred: Either invalid deployment or project ID. '
-            : '✗ An error occurred: ${e.message}',
-      );
+    if (deployment.buildId == null) {
+      logger.err('Build for $deploymentId has not started yet.');
       return ExitCode.software.code;
     }
 
+    final logs = await streamBuildLogs(
+      api: api,
+      orgId: validated.organization.id,
+      projectId: validated.project.id,
+      deploymentId: deploymentId,
+      buildId: deployment.buildId!,
+    );
+
     await printLogs(logger, logs);
 
     return ExitCode.success.code;

packages/globe_cli/lib/src/commands/deploy_command.dart

@@ -188,6 +188,7 @@ class DeployCommand extends BaseGlobeCommand {
             orgId: validated.organization.id,
             projectId: validated.project.id,
             deploymentId: deployment.id,
+            buildId: update.buildId!,
           );
 
           unawaited(

packages/globe_cli/lib/src/utils/api.dart

@@ -285,6 +285,65 @@ class GlobeApi {
       await http.delete(_buildUri(deleteTokenPath), headers: headers),
     )! as Map<String, Object?>;
   }
+
+  Future<String> getRealtimeToken({
+    required String orgId,
+    required String projectId,
+    required String deploymentId,
+    required String path,
+    Map<String, String>? headers,
+  }) {
+    final urlPath = [
+      'realtime',
+      'orgs',
+      orgId,
+      'projects',
+      projectId,
+      'deployments',
+      deploymentId,
+      path,
+      'tune',
+    ].join('/');
+
+    final uri = _buildUri('/$urlPath');
+    logger.detail('API Request: GET /$urlPath');
+
+    final request = http.Request('GET', uri);
+    request.headers.addAll(this.headers);
+    if (headers != null) {
+      request.headers.addAll(headers);
+    }
+
+    return request.send().then((response) async {
+      final json = jsonDecode(await response.stream.bytesToString())
+          as Map<String, Object?>;
+      if (response.statusCode == 200) {
+        return json['token']! as String;
+      } else {
+        throw ApiException._(response.statusCode, json['message']! as String);
+      }
+    });
+  }
+
+  Future<String> getBuildLogsToken({
+    required String orgId,
+    required String projectId,
+    required String deploymentId,
+    required String buildId,
+  }) {
+    requireAuth();
+
+    return getRealtimeToken(
+      orgId: orgId,
+      projectId: projectId,
+      deploymentId: deploymentId,
+      path: 'build-logs',
+      headers: {
+        'x-globe-build-id': buildId,
+        'x-globe-location': 'us-central1',
+      },
+    );
+  }
 }
 
 class Settings {
@@ -484,6 +543,7 @@ class Deployment {
     required this.active,
     required this.createdAt,
     required this.updatedAt,
+    required this.buildId,
   });
 
   factory Deployment.fromJson(Map<dynamic, dynamic> json) {
@@ -500,6 +560,7 @@ class Deployment {
         'active': final bool active,
         'createdAt': final String createdAt,
         'updatedAt': final String updatedAt,
+        'buildId': final String? buildId,
       } =>
         Deployment._(
           id: id,
@@ -519,6 +580,7 @@ class Deployment {
           active: active,
           createdAt: DateTime.parse(createdAt),
           updatedAt: DateTime.parse(updatedAt),
+          buildId: buildId,
         ),
       _ => throw const FormatException('Deployment'),
     };
@@ -535,6 +597,7 @@ class Deployment {
   final bool active;
   final DateTime createdAt;
   final DateTime updatedAt;
+  final String? buildId;
 }
 
 enum DeploymentState {

packages/globe_cli/lib/src/utils/logs.dart

@@ -1,7 +1,7 @@
 import 'dart:async';
 import 'dart:convert';
+import 'dart:io';
 
-import 'package:eventsource/eventsource.dart';
 import 'package:mason_logger/mason_logger.dart';
 
 import 'api.dart';
@@ -113,34 +113,36 @@ Future<Stream<BuildLogEvent>> streamBuildLogs({
   required String orgId,
   required String projectId,
   required String deploymentId,
+  required String buildId,
 }) async {
+  final host = Uri.parse(api.metadata.endpoint).host;
   final ctrl = StreamController<BuildLogEvent>.broadcast();
 
-  final es = await EventSource.connect(
-    '${api.metadata.endpoint}/api/orgs/$orgId/projects/$projectId/deployments/$deploymentId/build-logs',
-    headers: api.headers,
+  final buildLogsToken = await api.getBuildLogsToken(
+    orgId: orgId,
+    projectId: projectId,
+    deploymentId: deploymentId,
+    buildId: buildId,
   );
 
-  es.listen((e) {
-    if (e.data == null) {
-      return;
-    }
-
-    final json = jsonDecode(e.data!) as Map<String, dynamic>;
-
-    // The server emits a status event when a connection is established.
-    if (json['status'] != null) {
-      return;
-    }
+  final ws = await WebSocket.connect(
+    'wss://$host/api/realtime/orgs/$orgId/projects/$projectId/deployments/$deploymentId/$buildLogsToken',
+    headers: api.headers,
+  );
 
+  ws.listen((e) {
+    final json = jsonDecode(e as String) as Map<String, dynamic>;
     final event = BuildLogEvent.fromJson(json);
+
     ctrl.add(event);
   });
 
-  es.onError.listen((e) {
-    ctrl.add(ErrorBuildLogEvent(error: e.toString()));
+  unawaited(ws.done.then((_) => ctrl.close()));
+
+  ctrl.onCancel = () {
+    ws.close();
     ctrl.close();
-  });
+  };
 
   return ctrl.stream;
 }

packages/globe_cli/lib/src/utils/metadata.dart

@@ -15,7 +15,7 @@ class GlobeMetadata {
   );
 
   static const local = GlobeMetadata(
-    endpoint: 'http://localhost:8788',
+    endpoint: 'http://127.0.0.1:8788',
     isLocal: true,
   );
 

packages/globe_cli/pubspec.lock

@@ -193,14 +193,6 @@ packages:
       url: "https://pub.dev"
     source: hosted
     version: "2.0.5"
-  eventsource:
-    dependency: "direct main"
-    description:
-      name: eventsource
-      sha256: e21d60cd2320df8c8c303ebe11c80aa96b10928f411a12be31215062d8091be0
-      url: "https://pub.dev"
-    source: hosted
-    version: "0.4.0"
   file:
     dependency: "direct dev"
     description:
@@ -544,14 +536,6 @@ packages:
       url: "https://pub.dev"
     source: hosted
     version: "1.2.0"
-  sync:
-    dependency: transitive
-    description:
-      name: sync
-      sha256: f2ebb89eac969abb02b498562a35c4da63d6843396c4fe81948cd06a76845fce
-      url: "https://pub.dev"
-    source: hosted
-    version: "0.3.0"
   term_glyph:
     dependency: transitive
     description:

packages/globe_cli/pubspec.yaml

@@ -20,7 +20,6 @@ dependencies:
   pub_updater: ^0.3.0
   pubspec_parse: ^1.2.3
   pool: ^1.5.1
-  eventsource: ^0.4.0
 
 dev_dependencies:
   build_runner: ^2.4.4