RPC-less setup with Turbo and secondary connections with a custom channel

[raivil]

Hi,
First of all thanks for the awesome solutions!

Here's my situation
We configured the app to use anycable-pro (1.2.2-pro-96e71ad) and @anycable/turbo-stream on Rails 7.0.4.2 on a RPC-less infra. The app works as expected. This is awesome because we had previous experience with the gRPC server and it's way better to have one less component to manage.

Now we have a new demand to create a secondary websocket connection/client that will check user details (to simplify think of it as checking for the user session) on another host.
We have a multi-tenant app with multiple domains/subdomains that will authenticate on a centralized authentication server (OpenID Connect).
The clients on their subdomains (client1.example.com or client2.example.com) will "ping" host "auth.example.com" from time to time. One use case for it is to check if the user has a valid session on the auth server and logout the same user from all subdomains it might be logged.

A very basic implementation would be as follow:
app/channels/user_presence_channel.rb

class UserPresenceChannel < ApplicationCable::Channel
  def subscribed
    stream_from current_user
  end

  def unsubscribed
    # Any cleanup needed when channel is unsubscribed
  end

  def online?
    if current_user.present?  && current_user.any_other_specific_logic
      ActionCable.server.broadcast current_user, status: "valid"
    else
      ActionCable.server.broadcast current_user, status: "invalid"
    end
  end
end

class ApplicationCable::Connection < ActionCable::Connection::Base
  identified_by :current_user

  def connect
    self.current_user = find_verified_user
    reject_unauthorized_connection unless current_user
  end

  def find_verified_user (supports devise/warden)
end

app/javascript/channels/consumer.js

// Action Cable provides the framework to deal with WebSockets in Rails.
// You can generate new channels where WebSocket features live using the `bin/rails generate channel` command.

// import { createConsumer } from "@rails/actioncable"

// export default createConsumer()

// https://github.com/anycable/anycable-client#action-cable-compatibility-mode
import { createConsumer, fetchTokenFromHTML } from "@anycable/web";
import { ProtobufEncoder } from "@anycable/protobuf-encoder";

// code from https://github.com/anycable/anycable-client/blob/master/packages/web/index.js
const metaPrefixes = ["cable", "action-cable"];

const fetchMeta = (doc, key) => {
  for (const prefix of metaPrefixes) {
    const element = doc.head.querySelector(`meta[name='${prefix}-${key}']`);

    if (element) {
      return element.getAttribute("content");
    }
 }
};

// TODO: need to get the url according to the environment with a more robust way to generate it and get the token
// Here we have to point to "auth.example.com", so we need another way to build the websocket url with the token.
function getWebSocketURL() {
  const jid = fetchMeta(document, "url").split("cable?jid=")[1];
  return `ws://auth.example.com:8080/cable?jid=${jid}`;
}

// createConsumer accepts all the options available to createCable
// App is already configured for turbo. We do `createCable` with the same opts as below to use protobuf.
const consumer = createConsumer(getWebSocketURL(), {
  tokenRefresher: fetchTokenFromHTML(),
  protocol: "actioncable-v1-protobuf",
  encoder: new ProtobufEncoder(),
});
export default consumer;

/app/javascript/channels/user_presence_channel.js

import consumer from "./consumer";

consumer.subscriptions.create("UserPresenceChannel", {
  connected() {
    // Called when the subscription is ready for use on the server
    window.console.log("Connected.");
    setInterval(() => {
      this.online();
    }, 30000);
  },

  disconnected() {
    // Called when the subscription has been terminated by the server
    window.console.log("Disconnected.");
  },

  received(data) {
    // Called when there's incoming data on the websocket for this channel
    if (data.status === "invalid") {
      // eslint-disable-next-line i18n-text/no-en
      window.console.log("User session is invalid.");
    }
  },

  online() {
   // Implement a custom logic to check whatever the app needs.
    return this.perform("online?");
  },
});

This setup currently doesn't work, and requires the RPC server.
The connection doesn't get established and the following error shows on logs.

ws       |  ERROR 2023-02-22T20:00:00.912Z context=node enc=protobuf sid=21ReHr27b44zX0D5cZ2mn Subscribe error: grpc connection is not ready
ws       |   WARN 2023-02-22T20:00:00.912Z context=node enc=protobuf sid=21ReHr27b44zX0D5cZ2mn Failed to handle incoming message '!{"channel":"UserPresenceChannel"}' with error: grpc connection is not ready

Is it possible to still be RPC-less and be able to use custom channels and subscribers?
If so, how can this be implemented? Using JWT?
Can you provide any relevant information, details, ideas to help with this implementation?

Best regards,

[palkan]

Is it possible to still be RPC-less and be able to use custom channels and subscribers?

Since your channel's logic (UserPresenceChannel) is implemented within a Rails app, we cannot reach it w/o RPC.

There were some ideas on making it possible to implement simple channels right within anycable-go (for example, mruby commands). But I never had a change (and demand) to complete them.

Another option is to build a custom Go app on top of anycable-go and implement your channel in Go. There is no official documentation on how to do that yet. (Though we had build a handful of such apps at Evil Martians).

[raivil]

Thank you for the answer.
Seems like using the RPC is the way to go for now.

It should be able to run the rpc with the anycable-go server with a single command, at least on development environment.

Doing that I ended up with the following error. Not sure what is wrong. The error is only for the second websocket connection.
Any clues?

ws       | 2023-02-23 11:41:33.981809 D [48311:107340] [AnyCable sid=8O3SqHJIs7AczM2-VdIaB] ActionCable -- RPC Command: <AnyCable::CommandMessage: command: "subscribe", identifier: "{"channel":"UserPresenceChannel"}", connection_identifiers: "{"user":"Z2lkOi8vYWx1dW5vL1VzZXIvMQ"}", data: "", env: <AnyCable::Env: url: "http://auth.lvh.me:8080/cable?jid=eyJhbGciOiJIUzI1NiJ9.eyJleHQiOiJ7XCJ1c2VyXCI6XCJaMmxrT2k4dllXeDFkVzV2TDFWelpYSXZNUVwifSIsImV4cCI6MTY3NzE2Njg5M30.YbEO2JTeKdUod01lcn5VBVvbYx5bxv3hSJ7u0QidtNA", headers: {"REMOTE_ADDR"=>"127.0.0.1", "cookie"=>"_session_=a3b866a027c43724f477ba52990f6c47"}, cstate: {}, istate: {}>>
ws       | 2023-02-23 11:41:33.983036 D [48311:107340] [AnyCable sid=8O3SqHJIs7AczM2-VdIaB] ActionCable -- UserPresenceChannel is transmitting the subscription confirmation
ws       | 2023-02-23 11:41:33.983380 E [48311:107340 cli.rb:240] ActionCable -- Invalid argument for string field 'streams' (given NilClass).
ws       |  ERROR 2023-02-23T14:41:33.983Z context=node enc=protobuf sid=8O3SqHJIs7AczM2-VdIaB Subscribe error: Application error: Invalid argument for string field 'streams' (given NilClass).
ws       |   WARN 2023-02-23T14:41:33.983Z context=node enc=protobuf sid=8O3SqHJIs7AczM2-VdIaB Failed to handle incoming message '!{"channel":"UserPresenceChannel"}' with error: Application error: Invalid argument for string field 'streams' (given NilClass).

[raivil]

Turns out that channel's stream_from current_user was the cause since current_user is nil.
I replaced it with a random static name and it works.

This unblocks me to continue.
Thank you for the help!

ws       | 2023-02-23 14:27:05.067049 D [56462:107360] [AnyCable sid=ra4vKJtVWWfLm3SZS_S_1] ActionCable -- RPC Command: <AnyCable::CommandMessage: command: "subscribe", identifier: "{"channel":"UserPresenceChannel"}", connection_identifiers: "{"user":"Z2lkOi8vYWx1dW5vL1VzZXIvMQ"}", data: "", env: <AnyCable::Env: url: "http://auth.lvh.me:8080/cable?jid=eyJhbGciOiJIUzI1NiJ9.eyJleHQiOiJ7XCJ1c2VyXCI6XCJaMmxrT2k4dllXeDFkVzV2TDFWelpYSXZNUVwifSIsImV4cCI6MTY3NzE3NjgyNH0.jKcDMni4DgwsukY4f1Ui73VzarQ6F213bmqx3nwW-6U", headers: {"REMOTE_ADDR"=>"127.0.0.1", "cookie"=>"ahoy_visitor=946e9407-83fb-467e-92c6-e8d11abb68c1; __profilin=p%3Dt; ahoy_visitor=7e77e9fe-f00b-4d43-8c57-0de1e4a775d8; ahoy_visit=73acd44b-3b9c-4088-9591-7aa93f7a3277; ahoy_visit=3a0561ed-1fb5-4b3f-9f00-78329feffa33; _aluuno_session_v2=bfca2880ee31b0dd819dd92f381f6f09; d_aluuno_session_v2=c57e30cfd7a5a7ae6ea113887bdef869"}, cstate: {}, istate: {}>>
ws       | 2023-02-23 14:27:05.067618 D [56462:107360] [AnyCable sid=ra4vKJtVWWfLm3SZS_S_1] ActionCable -- UserPresenceChannel is transmitting the subscription confirmation
ws       |  DEBUG 2023-02-23T17:27:05.068Z context=rpc sid=ra4vKJtVWWfLm3SZS_S_1 Command response: status:SUCCESS streams:"user_presence_channel" transmissions:"{\"identifier\":\"{\\\"channel\\\":\\\"UserPresenceChannel\\\"}\",\"type\":\"confirm_subscription\"}" env:<> 
ws       |  DEBUG 2023-02-23T17:27:05.068Z context=node enc=protobuf sid=ra4vKJtVWWfLm3SZS_S_1 Subscribed to channel: {"channel":"UserPresenceChannel"}
ws       |  DEBUG 2023-02-23T17:27:05.068Z channel={"channel":"UserPresenceChannel"} context=hub sid=ra4vKJtVWWfLm3SZS_S_1 stream=user_presence_channel Subscribed

[palkan]

Invalid argument for string field 'streams' (given NilClass)

That's interesting; it seems that nil value for a stream name is allowed by Rails (and we just use what Rails gives us). Probably, because current_user is nil?

[raivil]

Yes. I believe that's the case.
I'm trying to figure out why the ApplicationCable::Connection/connect class is never called, thus not setting the current_user. That should explain the error.

[palkan]

why the ApplicationCable::Connection/connect class is never called

That's easy to understand: you're using JWT identification, which means you authenticating and identifying a connection via the token at WS side.

How do you generate a token? Do you pass a current user to it like this:

action_cable_with_jwt_meta_tag(current_user: current_user)

?

[raivil]

That's pretty much it!
(still learning here, these are my first days on action/anycable)
I've got the specific channel working with a secondary connection. Now it's a matter of implementing some business logic and deciding how to deploy the rpc server.
Thanks again for the help!