ArchEthic WebSocket client in Ruby

README.md

Experiment : connect an absinthe websocket in Ruby using async-websocket

In a process to port ArchEthic JS SDK (libjs) in Ruby, I'm stuck with asynchronous updates sent through WebSockets.

I reverse engineered the Absinthe Protocol to this point :

• Connect the Phoenix Server
• Send the subscription request
• Parse the responses
• Get one or two messages before being killed by hearthbeat

Questions :

• How to implement Heartbeat messages (every 30s when no other message has been sent)

Client -> [null,"9","phoenix","heartbeat",{}]
Server <- [null,"9","phoenix","phx_reply",{"response":{},"status":"ok"}]

• How to implement Subscription cancelling ?

• How to implement reconnection on error ?

  Server <- ["6","6","__absinthe__:control","phx_error",{}]  
  Client -> ["22","22","__absinthe__:control","phx_join",{}]
  Server <- ["22","22","__absinthe__:control","phx_reply",{"response":{},"status":"ok"}]

To start a local ArchEthic node

$ git clone https://github.com/archethic-foundation/archethic-node
$ cd archethic-node
$ docker build -t archethic-node .
# Edit docker-compose.yml to comment out node2  and node3
$ docker compose up
$ open http://localhost:4000

Useful links :

ArchEthic Node : https://github.com/archethic-foundation/archethic-node

ArchEthic libjs : https://github.com/archethic-foundation/libjs (see https://github.com/archethic-foundation/libjs/blob/master/lib/api.js#L288) keychain example is useful to spy websocket connections

1. start an ArchEthic node
2. in libjs : npm install
3. in example/keychain : npm install && npm start
4. open http://localhost:8080
5. check node endpoint (http://localhost:4000)
6. type anything as keychain seed
7. (start browser developer extensions, go to Network and filter WebSocket)
8. click "Create keychain"

Absinthe Websocket JS Client : https://github.com/absinthe-graphql/absinthe-socket

ae-oracle.rb

#!/bin/env ruby

require 'rubygems'
require 'bundler/setup'

require 'async/http'
require 'async/websocket'
require 'protocol/websocket/json_message'

# Well known ArchEthic entrypoints
AE_LOCAL   = 'http://localhost:4000'
AE_TESTNET = 'https://testnet.archethic.net'
AE_MAINNET = 'https://mainnet.archethic.net'

# GraphQL subscription to get Oracle Updates
ORACLE_SUBSCRIPTION = "subscription { oracleUpdate { timestamp, services { uco { eur, usd } } } }"

# For this experiment, we won't spam Mainnet/Testnet servers so go for the local node
http_endpoint = AE_LOCAL
puts "Connecting to #{http_endpoint}"

ws_endpoint = URI(AE_LOCAL)

# Build the WebSocket URI.
ws_endpoint.scheme = ws_endpoint.is_a?(URI::HTTPS) ? 'wss' : 'ws'
ws_endpoint = ws_endpoint.merge('/socket/websocket?vsn=2.0.0') # vsn=2.0.0 is mandatory !
puts "Websocket URL #{ws_endpoint}"

# Some Absinthe constants
ABSINTHE_CONTROL = '__absinthe__:control'
PHX_JOIN = 'phx_join'
PHX_REPLY = 'phx_reply'
GQL_DOC = "doc"

Async do |_task|
  endpoint = Async::HTTP::Endpoint.new(ws_endpoint)
  Async::WebSocket::Client.connect(endpoint) do |connection|

    # Each message is identified by a "ref". Each message should remind his join message "ref"
    join_ref = 0
    ref = 0

    # JOIN PHOENIX WEBSOCKET SERVER
    message = ::Protocol::WebSocket::JSONMessage.generate([join_ref.to_s, ref.to_s, ABSINTHE_CONTROL, PHX_JOIN, {}])
    message.send(connection)

    response = connection.read
    response = ::Protocol::WebSocket::JSONMessage.wrap(response).parse
    reply = response[4] if response.is_a?(Array) && response[0] == join_ref.to_s && response[1] == ref.to_s &&
      response[2] == ABSINTHE_CONTROL && response[3] == PHX_REPLY

    if reply.nil? || reply[:status] != 'ok'
      connection.close
      exit
    end

    puts "Joined Archethic WebSocket Server"

    # SUBSCRIBE TO ORACLE UPDATES
    ref += 1
    message = ::Protocol::WebSocket::JSONMessage.generate([join_ref.to_s, ref.to_s, ABSINTHE_CONTROL, GQL_DOC, { query: ORACLE_SUBSCRIPTION }])
    message.send(connection)

    response = connection.read
    response = ::Protocol::WebSocket::JSONMessage.wrap(response).parse
    reply = response[4] if response.is_a?(Array) && response[0] == join_ref.to_s && response[1] == ref.to_s &&
      response[2] == ABSINTHE_CONTROL && response[3] == PHX_REPLY

    if reply.nil? || reply[:status] != 'ok'
      connection.close
      exit
    end

    subscription_id = reply[:response][:subscriptionId]
    puts "Subscribed to Oracle Updated as #{subscription_id}"

    # NOW WAIT FOR ORACLE UPDATES...
    while message = connection.read
      message = ::Protocol::WebSocket::JSONMessage.wrap(message).parse
      puts message.inspect
    end
  end
end

Gemfile

source "https://rubygems.org"

gem "async-websocket"