Fix MQTT OAuth wildcard vhost matching and user cleanup on disconnect (#1745)

### WHAT is this pull request doing?
Fixes #1743

The MQTT connection factory used `user.permissions.has_key?(vhost)` for
vhost permission checks, which only does exact hash key lookup. The AMQP
connection factory uses `user.find_permission(vhost)` which supports
wildcard matching (e.g. a scope with vhost `*` matching `/`). This
caused MQTT OAuth connections to fail when JWT token scopes use wildcard
vhosts, even though the same token worked for AMQP.

Also adds `OAuthUser#cleanup` on MQTT client disconnect to prevent
expiration monitoring fiber leaks, consistent with the AMQP client.

  ### HOW can this pull request be tested?
Run with: `make test SPEC=spec/mqtt/oauth_spec.cr`

Author: kickster97

spec/mqtt/oauth_spec.cr

@@ -0,0 +1,103 @@
+require "./spec_helper"
+require "../oauth_spec"
+
+module MqttSpecs
+  extend MqttHelpers
+  extend MqttMatchers
+
+  describe "MQTT OAuth" do
+    it "allows connection with wildcard vhost permissions" do
+      with_server do |server|
+        reader, writer = IO.pipe
+        mqtt_io = MQTT::Protocol::IO.new(reader)
+        conn_info = LavinMQ::ConnectionInfo.local
+
+        # OAuthUser with wildcard vhost "*" should match default vhost "/"
+        permissions = {"*" => {config: /.*/, read: /.*/, write: /.*/}}
+        user = OAuthUserHelper.create_user(RoughTime.utc + 1.hour, permissions)
+
+        broker = server.@mqtt_brokers.@brokers["/"]
+        packet = MQTT::Protocol::Connect.new(
+          client_id: "oauth-wildcard-test",
+          clean_session: true,
+          keepalive: 60u16,
+          username: "testuser",
+          password: nil,
+          will: nil,
+        )
+
+        broker.add_client(mqtt_io, conn_info, user, packet)
+
+        sleep 100.milliseconds
+
+        # Client should be connected
+        reader.closed?.should be_false
+      ensure
+        reader.try &.close
+        writer.try &.close
+      end
+    end
+
+    it "allows connection with exact vhost permissions" do
+      with_server do |server|
+        reader, writer = IO.pipe
+        mqtt_io = MQTT::Protocol::IO.new(reader)
+        conn_info = LavinMQ::ConnectionInfo.local
+
+        permissions = {"/" => {config: /.*/, read: /.*/, write: /.*/}}
+        user = OAuthUserHelper.create_user(RoughTime.utc + 1.hour, permissions)
+
+        broker = server.@mqtt_brokers.@brokers["/"]
+        packet = MQTT::Protocol::Connect.new(
+          client_id: "oauth-exact-test",
+          clean_session: true,
+          keepalive: 60u16,
+          username: "testuser",
+          password: nil,
+          will: nil,
+        )
+
+        broker.add_client(mqtt_io, conn_info, user, packet)
+
+        sleep 100.milliseconds
+
+        reader.closed?.should be_false
+      ensure
+        reader.try &.close
+        writer.try &.close
+      end
+    end
+
+    it "cleans up OAuthUser on disconnect" do
+      with_server do |server|
+        reader, writer = IO.pipe
+        mqtt_io = MQTT::Protocol::IO.new(reader)
+        conn_info = LavinMQ::ConnectionInfo.local
+
+        permissions = {"/" => {config: /.*/, read: /.*/, write: /.*/}}
+        user = OAuthUserHelper.create_user(RoughTime.utc + 1.hour, permissions)
+
+        broker = server.@mqtt_brokers.@brokers["/"]
+        packet = MQTT::Protocol::Connect.new(
+          client_id: "oauth-cleanup-test",
+          clean_session: true,
+          keepalive: 60u16,
+          username: "testuser",
+          password: nil,
+          will: nil,
+        )
+
+        broker.add_client(mqtt_io, conn_info, user, packet)
+
+        # Close the socket to trigger disconnect and cleanup
+        reader.close
+        sleep 200.milliseconds
+
+        # The OAuthUser's token_updated channel should be closed by cleanup
+        user.@token_updated.closed?.should be_true
+      ensure
+        writer.try &.close
+      end
+    end
+  end
+end

src/lavinmq/mqtt/client.cr

@@ -82,6 +82,10 @@ module LavinMQ
         publish_will
       ensure
         @broker.remove_client(self)
+        case user = @user
+        when Auth::OAuthUser
+          user.cleanup
+        end
         @waitgroup.done
         close_socket
         @log.info { "Connection disconnected for user=#{@user.name} duration=#{duration}" }

src/lavinmq/mqtt/connection_factory.cr

@@ -66,8 +66,7 @@ module LavinMQ
 
         user = @authenticator.authenticate(context)
         return unless user
-        has_vhost_permissions = user.try &.permissions.has_key?(vhost)
-        return unless has_vhost_permissions
+        return unless user.find_permission(vhost)
         broker = @brokers[vhost]?
         return unless broker