feat: logs

app/routes/logs.ts

@@ -0,0 +1,34 @@
+import { json } from "@remix-run/node";
+import { SignalRSocket } from "~/signalr/signalr-socket";
+import { w3cwebsocket } from "websocket";
+import type { Params } from "@remix-run/react";
+
+(globalThis as any).WebSocket = w3cwebsocket;
+
+export async function loader({
+  params,
+}: {
+  params: Params<"owner" | "repo" | "commit_hash" | "check_id">;
+}) {
+  const instance = SignalRSocket.getInstance("WatchRunStepsProgressAsync");
+  instance.startOrContinueStreaming(
+    `https://github.com/${params.owner}/${params.repo}/commit/${params.commit_hash}/checks/${params.check_id}/live_logs`
+  );
+  let _resolve: (arg0: Event) => void;
+  const res = new Promise<Event>((resolve) => (_resolve = resolve));
+
+  instance.addEventListener("socketDidReceiveMessage", (event) =>
+    _resolve(event)
+  );
+
+  const seconds = 50;
+  return json({
+    hello: await Promise.race([
+      res,
+      sleep(seconds).then(() => ({ timedOut: `failed in ${seconds} seconds` })),
+    ]),
+  });
+}
+
+const sleep = (seconds: number) =>
+  new Promise<void>((resolve) => setTimeout(resolve, seconds * 1000));

app/signalr/fetch.ts

@@ -0,0 +1,19 @@
+export async function fetchJson<T>(url: string): Promise<T | null> {
+  try {
+    const response = await fetch(url, {
+      // credentials: 'same-origin',
+      headers: {
+        cookie: process.env.GITHUB_COOKIE!,
+        Accept: "application/json",
+        "X-Requested-With": "XMLHttpRequest",
+      },
+    });
+    if (response.ok) {
+      return await response.json();
+    }
+  } catch (e) {
+    console.error(e);
+    // Network failure
+  }
+  return null;
+}

app/signalr/signalr-socket.ts

@@ -0,0 +1,201 @@
+import type { Socket } from "@github/stable-socket";
+import { StableSocket } from "@github/stable-socket";
+import { fetchJson } from "./fetch";
+
+interface AuthenticatedURLResponse {
+  data: {
+    authenticated_url: string;
+  };
+}
+
+interface WsUrlLookupData {
+  logStreamWebSocketUrl: string;
+}
+
+export enum MessageType {
+  Invocation = 1,
+  StreamItem = 2,
+  Completion = 3,
+  StreamInvocation = 4,
+  CancelInvocation = 5,
+  Ping = 6,
+  Close = 7,
+}
+
+// https://github.com/dotnet/aspnetcore/blob/main/src/SignalR/docs/specs/HubProtocol.md#overview
+type Message = HandshakeRequest | Invocation;
+
+interface HandshakeRequest {
+  protocol: string;
+  version: number;
+}
+
+interface Invocation {
+  type: number;
+  target: string;
+  arguments: unknown[];
+}
+
+export interface SocketData<T> {
+  target: string;
+  type: number;
+  arguments: T;
+}
+
+export class SignalRSocket extends EventTarget {
+  public static readonly RECORD_SEPARATOR = String.fromCharCode(0x1e);
+  // https://developer.mozilla.org/en-US/docs/Web/API/CloseEvent
+  public static readonly ABNORMAL_CLOSURE = 1006;
+  public static readonly POLICY_VIOLATION = 1008;
+  static #instances = new Map<string, SignalRSocket>();
+
+  private globalSocket!: Socket | null;
+  private socketUrl!: string;
+  private isStreaming = false;
+  private xhrConnectionUrl!: string;
+  private shouldTryAgainWithNewSocket = false;
+
+  public static getInstance(target: string): SignalRSocket {
+    if (!SignalRSocket.#instances.has(target)) {
+      SignalRSocket.#instances.set(target, new SignalRSocket(target));
+    }
+    return SignalRSocket.#instances.get(target)!;
+  }
+
+  constructor(private target: string) {
+    super();
+  }
+
+  public socketDidReceiveMessage(socket: Socket, message: string): void {
+    const record = message.split(SignalRSocket.RECORD_SEPARATOR)[0]!;
+    const data = JSON.parse(record) as SocketData<unknown>;
+
+    // this.dispatchEvent(new CustomEvent('socketDidReceiveMessage', {detail: data}))
+
+    if (data.type === MessageType.Close) {
+      socket.close();
+    } else if (data.type === MessageType.Ping) {
+      console.log("ping");
+    } else {
+      console.log({ data });
+    }
+  }
+
+  // Public methods
+  public socketDidOpen() {
+    this.handshake();
+  }
+
+  public socketDidClose() {
+    // Do nothing
+  }
+
+  public socketDidFinish() {
+    this.globalSocket = null;
+
+    if (this.shouldTryAgainWithNewSocket) {
+      this.shouldTryAgainWithNewSocket = false;
+      this.startOrContinueStreaming(this.xhrConnectionUrl, true);
+    }
+  }
+
+  public socketShouldRetry(socket: Socket, code: number): boolean {
+    if (code === SignalRSocket.ABNORMAL_CLOSURE) {
+      /**
+       * AFD (Azure Front Door) abnormally closed the socket. This can happen after the websocket max time limit
+       * has been reached which is configured to 10 minutes for Actions. An abnormal closure can also sometimes
+       * randomly happen before the max time limit has been reached which stops incoming logs.
+       *
+       * Retrying with the existing socket is futile, but it is safe to discard the existing socket and create a new one.
+       */
+      this.shouldTryAgainWithNewSocket = true;
+      return false;
+    }
+    return !this.isFatalStatusCode(code);
+  }
+
+  public async startOrContinueStreaming(
+    xhrConnectionUrl: string,
+    retryingAbnormalClosure = false
+  ) {
+    if (!this.xhrConnectionUrl) this.xhrConnectionUrl = xhrConnectionUrl;
+
+    if (this.isStreaming && !retryingAbnormalClosure) return;
+    this.isStreaming = true;
+
+    await this.tryStartNewStream(this.xhrConnectionUrl);
+  }
+
+  public closeExistingSocket() {
+    if (this.globalSocket) {
+      this.globalSocket.close();
+      this.globalSocket = null;
+    }
+    return;
+  }
+
+  public onStreamFailure() {
+    this.dispatchEvent(new CustomEvent("onStreamFailure"));
+  }
+
+  // Private methods
+  private async tryStartNewStream(xhrConnectionUrl: string) {
+    try {
+      const socketUrl = await this.retrieveSocketURL(xhrConnectionUrl);
+      if (!socketUrl) return;
+      this.socketUrl = socketUrl;
+
+      this.globalSocket = new StableSocket(this.socketUrl, this, {
+        timeout: 4000,
+        attempts: 10,
+      });
+      await this.globalSocket.open();
+    } catch (e) {
+      console.log({ e });
+      this.onStreamFailure();
+      throw e;
+    }
+  }
+
+  private handshake() {
+    const { searchParams } = new URL(this.socketUrl); //, window.location.origin)
+    const tenantId = searchParams.get("tenantId");
+    const runId = searchParams.get("runId");
+    if (tenantId && runId) {
+      const protocolPayload: HandshakeRequest = {
+        protocol: "json",
+        version: 1,
+      };
+      this.sendPayload(protocolPayload);
+
+      const targetPayload: Invocation = {
+        arguments: [tenantId, +runId],
+        target: this.target,
+        type: MessageType.Invocation,
+      };
+      this.sendPayload(targetPayload);
+    }
+  }
+
+  private retrieveSocketURL = async (
+    xhrUrl: string
+  ): Promise<string | null> => {
+    const authResult = await fetchJson<AuthenticatedURLResponse>(xhrUrl);
+    const authenticatedUrl = authResult?.data["authenticated_url"];
+    if (!authenticatedUrl) return null;
+    const wsResult = await fetchJson<WsUrlLookupData>(authenticatedUrl);
+    return wsResult?.logStreamWebSocketUrl ?? null;
+  };
+
+  private isFatalStatusCode(code: number): boolean {
+    return code === SignalRSocket.POLICY_VIOLATION;
+  }
+
+  private sendPayload(payload: Message) {
+    if (this.globalSocket) {
+      this.globalSocket.send(
+        JSON.stringify(payload) + SignalRSocket.RECORD_SEPARATOR
+      );
+    }
+  }
+}

package.json

@@ -12,6 +12,10 @@
     "pull": "vc env pull .env"
   },
   "dependencies": {
+    "@github/catalyst": "^1.6.0",
+    "@github/hotkey": "^2.0.1",
+    "@github/jtml": "^0.5.0",
+    "@github/stable-socket": "^1.1.0",
     "@octokit/auth-app": "^6.0.0",
     "@octokit/auth-oauth-app": "^7.0.1",
     "@octokit/plugin-throttling": "^8.0.0",
@@ -29,18 +33,22 @@
     "@types/async": "^3.2.18",
     "@types/humanize-duration": "^3.27.1",
     "@types/lodash": "^4.14.191",
+    "@types/websocket": "^1.0.5",
     "@vercel/analytics": "^1.0.0",
     "@vercel/kv": "^1.0.0",
     "@vercel/node": "^3.0.0",
     "adm-zip": "^0.5.10",
     "ansicolor": "^2.0.1",
     "async": "^3.2.4",
+    "bulma": "^0.9.4",
     "debug": "^4.3.4",
+    "delegated-events": "^1.1.2",
     "github-syntax-dark": "^0.5.0",
     "github-syntax-light": "^0.5.0",
     "graphql": "^16.6.0",
     "graphql-tag": "^2.12.6",
     "humanize-duration": "^3.28.0",
+    "lit-html": "^2.6.1",
     "lodash": "^4.17.21",
     "react": "^18.2.0",
     "react-dom": "^18.2.0",
@@ -50,8 +58,10 @@
     "remix-auth-github": "^1.3.0",
     "remix-auth-oauth2": "^1.11.0",
     "remix-island": "^0.1.2",
+    "selector-observer": "^2.1.6",
     "styled-components": "^5.0.0",
-    "type-fest": "^4.0.0"
+    "type-fest": "^4.0.0",
+    "websocket": "^1.0.34"
   },
   "devDependencies": {
     "@graphql-codegen/cli": "^5.0.0",

remix.config.js

@@ -10,6 +10,7 @@ module.exports = {
   // assetsBuildDirectory: "public/build",
   // serverBuildPath: "api/index.js",
   // publicPath: "/build/",
+  serverDependenciesToBundle: ["@github/catalyst", "@github/stable-socket"],
   serverModuleFormat: 'cjs',
   future: {
   },