Stream Zaptec Service Bus messages

lib/stekker_zaptec.rb

@@ -10,9 +10,12 @@
 require "zaptec/errors"
 require "zaptec/installation"
 require "zaptec/installation_hierarchy"
+require "zaptec/messaging_connection_details"
+require "zaptec/messaging_subscription"
 require "zaptec/meter_reading"
 require "zaptec/null_encryptor"
 require "zaptec/state"
+require "zaptec/user_group"
 require "zaptec/version"
 
 module Zaptec

lib/zaptec/client.rb

@@ -110,6 +110,19 @@ def update_charger(charger_id, **attributes)
       post("/api/chargers/#{charger_id}/update", body: attributes)
     end
 
+    def user_groups
+      response = get("/api/userGroups")
+      data = response.body
+      data = data.fetch("Data") if data.is_a?(Hash) && data.key?("Data")
+      data.map { |attrs| UserGroup.new(attrs) }
+    end
+
+    # https://docs.zaptec.com/docs/zaptec-service-bus-subscriptions
+    def messaging_connection_details(user_group_id)
+      get("/api/userGroups/#{user_group_id}/messagingConnectionDetails")
+        .then { |response| MessagingConnectionDetails.new(response.body) }
+    end
+
     def pause_charging(charger_id) = send_command(charger_id, :StopChargingFinal)
 
     def resume_charging(charger_id) = send_command(charger_id, :ResumeCharging)

lib/zaptec/errors.rb

@@ -3,14 +3,31 @@ module Errors
     class Base < StandardError; end
     class ParameterMissing < Base; end
     class Forbidden < Base; end
+
     class RequestFailed < Base
       attr_reader :response
 
       def initialize(message, response = nil)
         @response = response
-        super(message)
+        super(format_message(message, response))
+      end
+
+      private
+
+      def format_message(message, response)
+        body = extract_body(response)
+        body.present? ? "#{message}: #{body}" : message
+      end
+
+      def extract_body(response)
+        case response
+        when nil then nil
+        when Hash then response[:body]
+        else response.respond_to?(:body) ? response.body : nil
+        end
       end
     end
+
     class AuthorizationFailed < Base; end
   end
 end

lib/zaptec/messaging_connection_details.rb

@@ -0,0 +1,15 @@
+module Zaptec
+  class MessagingConnectionDetails
+    def initialize(data)
+      @data = data.deep_symbolize_keys
+    end
+
+    def host = @data.fetch(:Host)
+    def port = @data.fetch(:Port)
+    def use_ssl = @data.fetch(:UseSSL)
+    def username = @data.fetch(:Username)
+    def password = @data.fetch(:Password)
+    def topic = @data.fetch(:Topic)
+    def subscription = @data.fetch(:Subscription)
+  end
+end

lib/zaptec/messaging_subscription.rb

@@ -0,0 +1,164 @@
+require "base64"
+require "json"
+require "net/http"
+require "openssl"
+require "uri"
+
+module Zaptec
+  class MessagingSubscription
+    Message = Data.define(
+      :device_id,
+      :device_type,
+      :state_id,
+      :state_name,
+      :timestamp,
+      :value,
+      :raw,
+      :lock_location,
+    ) do
+      def initialize(lock_location: nil, **)
+        super
+      end
+    end
+
+    TOKEN_TTL = 1.hour
+    TOKEN_REFRESH_MARGIN = 5.minutes
+
+    def initialize(connection_details:)
+      @connection_details = connection_details
+    end
+
+    def each_message(mode: :consume, timeout: 30)
+      return enum_for(:each_message, mode:, timeout:) unless block_given?
+
+      loop do
+        message = fetch_one(mode:, timeout:)
+        yield message if message
+      end
+    end
+
+    def complete(message)
+      raise ArgumentError, "Message has no lock; was it received in peek mode?" if message.lock_location.nil?
+
+      response = http.request(lock_request(Net::HTTP::Delete, message.lock_location))
+      handle_lock_response(response)
+    end
+
+    private
+
+    attr_reader :connection_details
+
+    def fetch_one(mode:, timeout:)
+      response = http.request(message_request(mode:, timeout:))
+
+      case response
+      when Net::HTTPNoContent
+        nil
+      when Net::HTTPSuccess
+        parse_message(response.body.to_s, lock_location: response["Location"])
+      when Net::HTTPUnauthorized
+        invalidate_token!
+        raise Errors::AuthorizationFailed
+      else
+        raise Errors::RequestFailed.new("#{response.code} #{response.message}: #{response.body}", response)
+      end
+    end
+
+    def message_request(mode:, timeout:)
+      klass = mode == :peek ? Net::HTTP::Post : Net::HTTP::Delete
+      klass.new(message_path(timeout)).tap do |req|
+        req["Authorization"] = sas_token
+        req["Content-Length"] = "0"
+      end
+    end
+
+    def lock_request(klass, lock_location)
+      uri = URI(lock_location)
+      klass.new(uri.request_uri).tap do |req|
+        req["Authorization"] = sas_token
+        req["Content-Length"] = "0"
+      end
+    end
+
+    def handle_lock_response(response)
+      case response
+      when Net::HTTPSuccess
+        nil
+      when Net::HTTPUnauthorized
+        invalidate_token!
+        raise Errors::AuthorizationFailed
+      else
+        raise Errors::RequestFailed.new("#{response.code} #{response.message}: #{response.body}", response)
+      end
+    end
+
+    def message_path(timeout)
+      "/#{connection_details.topic}/subscriptions/#{connection_details.subscription}" \
+        "/messages/head?timeout=#{timeout}"
+    end
+
+    def parse_message(body, lock_location:)
+      start_index = body.index("{")
+      end_index = body.rindex("}")
+      return nil unless start_index && end_index && end_index > start_index
+
+      raw = JSON.parse(body[start_index..end_index])
+      state_id = raw["StateId"]
+      device_type = raw["DeviceType"]
+
+      Message.new(
+        device_id: raw["DeviceId"],
+        device_type:,
+        state_id:,
+        state_name: state_name_for(state_id, device_type),
+        timestamp: raw["Timestamp"],
+        value: raw["ValueAsString"],
+        raw:,
+        lock_location:,
+      )
+    rescue JSON::ParserError
+      nil
+    end
+
+    def state_name_for(state_id, device_type)
+      return unless state_id && device_type
+
+      Constants.observation_state_id_to_name(state_id:, device_type:)
+    end
+
+    def sas_token
+      now = Time.now.to_i
+      if @token.nil? || now > @token_expiry - TOKEN_REFRESH_MARGIN.to_i
+        @token_expiry = now + TOKEN_TTL.to_i
+        @token = build_sas_token(expiry: @token_expiry)
+      end
+      @token
+    end
+
+    def build_sas_token(expiry:)
+      encoded_uri = URI.encode_www_form_component("https://#{connection_details.host}/")
+      signature = Base64.strict_encode64(
+        OpenSSL::HMAC.digest("sha256", connection_details.password, "#{encoded_uri}\n#{expiry}"),
+      )
+
+      "SharedAccessSignature " \
+        "sr=#{encoded_uri}" \
+        "&sig=#{URI.encode_www_form_component(signature)}" \
+        "&se=#{expiry}" \
+        "&skn=#{connection_details.username}"
+    end
+
+    def invalidate_token!
+      @token = nil
+      @token_expiry = nil
+    end
+
+    def http
+      @http ||=
+        Net::HTTP.new(connection_details.host, 443).tap do |conn|
+          conn.use_ssl = true
+          conn.read_timeout = 90
+        end
+    end
+  end
+end

lib/zaptec/user_group.rb

@@ -0,0 +1,10 @@
+module Zaptec
+  class UserGroup
+    def initialize(data)
+      @data = data.deep_symbolize_keys
+    end
+
+    def id = @data.fetch(:Id) { @data.fetch(:id) }
+    def name = @data[:Name] || @data[:name]
+  end
+end

spec/zaptec/client_spec.rb

@@ -651,6 +651,77 @@
     end
   end
 
+  describe "#user_groups" do
+    it "returns the user groups accessible to the authenticated user" do
+      WebMock::API
+        .stub_request(:get, "https://api.zaptec.com/api/userGroups")
+        .with(headers: { "Authorization" => "Bearer T123" })
+        .to_return(
+          body: {
+            Data: [
+              { Id: "g1", Name: "Acme" },
+              { Id: "g2", Name: "Beta" },
+            ],
+          }.to_json,
+          headers: { "Content-Type": "application/json" },
+        )
+
+      token_cache = build_token_cache("T123")
+      client = Zaptec::Client.new(username: "zap", password: "tec", token_cache:)
+
+      expect(client.user_groups.map(&:id)).to eq %w[g1 g2]
+      expect(client.user_groups.map(&:name)).to eq %w[Acme Beta]
+    end
+
+    it "supports an unwrapped array response" do
+      WebMock::API
+        .stub_request(:get, "https://api.zaptec.com/api/userGroups")
+        .with(headers: { "Authorization" => "Bearer T123" })
+        .to_return(
+          body: [{ Id: "g1", Name: "Acme" }].to_json,
+          headers: { "Content-Type": "application/json" },
+        )
+
+      token_cache = build_token_cache("T123")
+      client = Zaptec::Client.new(username: "zap", password: "tec", token_cache:)
+
+      expect(client.user_groups.map(&:id)).to eq %w[g1]
+    end
+  end
+
+  describe "#messaging_connection_details" do
+    it "fetches Service Bus connection details for a user group" do
+      WebMock::API
+        .stub_request(:get, "https://api.zaptec.com/api/userGroups/g123/messagingConnectionDetails")
+        .with(headers: { "Authorization" => "Bearer T123" })
+        .to_return(
+          body: {
+            Type: 0,
+            Host: "sb.example.com",
+            Port: 5671,
+            UseSSL: true,
+            Username: "usergroup_g123",
+            Password: "secret",
+            Topic: "usergroup_g123",
+            Subscription: "default",
+          }.to_json,
+          headers: { "Content-Type": "application/json" },
+        )
+
+      token_cache = build_token_cache("T123")
+      client = Zaptec::Client.new(username: "zap", password: "tec", token_cache:)
+
+      expect(client.messaging_connection_details("g123"))
+        .to have_attributes(
+          host: "sb.example.com",
+          username: "usergroup_g123",
+          password: "secret",
+          topic: "usergroup_g123",
+          subscription: "default",
+        )
+    end
+  end
+
   private
 
   def chargers_example

spec/zaptec/errors/request_failed_spec.rb

@@ -0,0 +1,35 @@
+RSpec.describe Zaptec::Errors::RequestFailed do
+  it "appends the body from a Faraday response hash to the message" do
+    error = described_class.new("Request returned status 503", { body: "Service Unavailable" })
+
+    expect(error.message).to eq("Request returned status 503: Service Unavailable")
+  end
+
+  it "appends the body from a Net::HTTP response object to the message" do
+    response = instance_double(Net::HTTPResponse, body: "<html>nope</html>")
+
+    error = described_class.new("503 Service Unavailable", response)
+
+    expect(error.message).to eq("503 Service Unavailable: <html>nope</html>")
+  end
+
+  it "omits the body separator when no response is given" do
+    error = described_class.new("something broke")
+
+    expect(error.message).to eq("something broke")
+  end
+
+  it "omits the body separator when the body is blank" do
+    error = described_class.new("Request returned status 500", { body: "" })
+
+    expect(error.message).to eq("Request returned status 500")
+  end
+
+  it "exposes the original response object" do
+    response = { status: 503, body: "down" }
+
+    error = described_class.new("boom", response)
+
+    expect(error.response).to eq(response)
+  end
+end

spec/zaptec/messaging_connection_details_spec.rb

@@ -0,0 +1,24 @@
+RSpec.describe Zaptec::MessagingConnectionDetails do
+  it "exposes the connection settings" do
+    details = described_class.new(
+      "Type" => 0,
+      "Host" => "zap-p-installations-sbus.servicebus.windows.net",
+      "Port" => 5671,
+      "UseSSL" => true,
+      "Username" => "usergroup_abc",
+      "Password" => "secret",
+      "Topic" => "usergroup_abc",
+      "Subscription" => "default",
+    )
+
+    expect(details).to have_attributes(
+      host: "zap-p-installations-sbus.servicebus.windows.net",
+      port: 5671,
+      use_ssl: true,
+      username: "usergroup_abc",
+      password: "secret",
+      topic: "usergroup_abc",
+      subscription: "default",
+    )
+  end
+end