Disconnect MQTT clients on OAuth token expiration (#1746)

### WHAT is this pull request doing?

Fixes #1744

The AMQP client sets up an `on_expiration` callback that closes the
connection when an OAuth token expires. The MQTT client had no
equivalent, so MQTT clients with expired tokens remained connected
indefinitely.

Adds the same `on_expiration` callback to the MQTT client, closing the
connection with reason "token expired" when the token lifetime runs out.

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

Author: kickster97

spec/mqtt/oauth_expiration_spec.cr

@@ -0,0 +1,74 @@
+require "./spec_helper"
+require "../oauth_spec"
+
+module MqttSpecs
+  extend MqttHelpers
+  extend MqttMatchers
+
+  describe "MQTT OAuth token expiration" do
+    it "disconnects client when token expires" do
+      with_server do |server|
+        # Create a pipe pair to simulate the MQTT socket
+        reader, writer = IO.pipe
+        mqtt_io = MQTT::Protocol::IO.new(reader)
+        conn_info = LavinMQ::ConnectionInfo.local
+
+        # Create OAuthUser with a very short-lived token
+        permissions = {"/" => {config: /.*/, read: /.*/, write: /.*/}}
+        user = OAuthUserHelper.create_user(RoughTime.utc + 50.milliseconds, permissions)
+
+        broker = server.@mqtt_brokers.@brokers["/"]
+        packet = MQTT::Protocol::Connect.new(
+          client_id: "oauth-expiry-test",
+          clean_session: true,
+          keepalive: 60u16,
+          username: "testuser",
+          password: nil,
+          will: nil,
+        )
+
+        broker.add_client(mqtt_io, conn_info, user, packet)
+
+        # Client should be connected
+        reader.closed?.should be_false
+
+        # Wait for token to expire and the on_expiration callback to fire
+        wait_for { reader.closed? }.should be_true
+      ensure
+        writer.try &.close
+      end
+    end
+
+    it "does not disconnect client when token is still valid" do
+      with_server do |server|
+        reader, writer = IO.pipe
+        mqtt_io = MQTT::Protocol::IO.new(reader)
+        conn_info = LavinMQ::ConnectionInfo.local
+
+        # Create OAuthUser with a long-lived token
+        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-valid-test",
+          clean_session: true,
+          keepalive: 60u16,
+          username: "testuser",
+          password: nil,
+          will: nil,
+        )
+
+        broker.add_client(mqtt_io, conn_info, user, packet)
+
+        sleep 50.milliseconds
+
+        # Client should still be connected
+        reader.closed?.should be_false
+      ensure
+        reader.try &.close
+        writer.try &.close
+      end
+    end
+  end
+end

src/lavinmq/mqtt/client.cr

@@ -41,6 +41,12 @@ module LavinMQ
         @log = Logger.new(Log, metadata)
         @log.info { "Connection established for user=#{@user.name}" }
         spawn read_loop, name: "MQTT read_loop #{@connection_info.remote_address}"
+        case user = @user
+        when Auth::OAuthUser
+          user.on_expiration do
+            close("token expired")
+          end
+        end
       end
 
       def client_name