MQTT Integration

@Configuration
class MqttConfig(
    @param:Value("\${spring.mqtt.broker.url}")
    private val brokerUrl: String,  // e.g., wss://mqtt.example.com:8084/mqtt

    @param:Value("\${spring.mqtt.username}")
    private val username: String,

    @param:Value("\${spring.mqtt.password}")
    private val password: String,

    @param:Value("\${spring.mqtt.client.id-prefix:greenhouse-api}")
    private val clientIdPrefix: String,

    @param:Value("\${spring.mqtt.client.clean-session:false}")
    private val cleanSession: Boolean,

    @param:Value("\${spring.mqtt.client.connection-timeout:10}")
    private val connectionTimeout: Int,

    @param:Value("\${spring.mqtt.client.keep-alive-interval:60}")
    private val keepAliveInterval: Int,

    @param:Value("\${spring.mqtt.client.automatic-reconnect:true}")
    private val automaticReconnect: Boolean,

    // ... listener dependencies
) {
    /**
     * Factory for creating MQTT clients configured per official documentation
     * https://docs.spring.io/spring-integration/reference/mqtt.html
     */
    @Bean
    fun mqttClientFactory(): MqttPahoClientFactory {
        val factory = DefaultMqttPahoClientFactory()
        val options = MqttConnectOptions()

        // Configure connection options
        options.serverURIs = arrayOf(brokerUrl)
        options.userName = username
        options.password = password.toCharArray()
        options.isCleanSession = cleanSession
        options.connectionTimeout = connectionTimeout
        options.keepAliveInterval = keepAliveInterval
        options.isAutomaticReconnect = automaticReconnect

        factory.connectionOptions = options
        return factory
    }
}

spring:
  mqtt:
    broker:
      url: ${MQTT_BROKER_URL:wss://mqtt.invernaderos.com:8084/mqtt}
    username: ${MQTT_USERNAME:api-client}
    password: ${MQTT_PASSWORD}
    client:
      id-prefix: greenhouse-api
      clean-session: false  # Persist session state
      connection-timeout: 10  # seconds
      keep-alive-interval: 60  # seconds
      automatic-reconnect: true
    topics:
      greenhouse-multi-tenant: GREENHOUSE/+  # Wildcard for all tenants
      sensors-pattern: greenhouse/+/sensors/#
      actuators-pattern: greenhouse/+/actuators/#
      system-events: system/events/#
    qos:
      default: 0  # At most once (fire and forget)

@Bean
fun mqttInbound(): MessageProducer {
    // Generate a unique client ID for the inbound adapter
    val clientId = "$clientIdPrefix-inbound-${UUID.randomUUID()}"

    // Configure topics to subscribe
    val topics = arrayOf(
        "GREENHOUSE",                    // Legacy topic (backward compatibility)
        greenhouseMultiTenantPattern,    // Multi-tenant: GREENHOUSE/+ (wildcard)
        sensorsTopicPattern,             // greenhouse/+/sensors/#
        actuatorsTopicPattern,           // greenhouse/+/actuators/#
        systemEventsTopicPattern         // system/events/#
    )

    // Create the adapter with client ID, factory, and topics
    val adapter = MqttPahoMessageDrivenChannelAdapter(
        clientId,
        mqttClientFactory(),
        *topics
    )

    // Configure the adapter
    adapter.setCompletionTimeout(5000)
    adapter.setConverter(DefaultPahoMessageConverter())
    adapter.setQos(defaultQos)  // QoS 0 by default
    adapter.outputChannel = mqttInputChannel()

    return adapter
}

@Bean
@ServiceActivator(inputChannel = "mqttInputChannel")
fun mqttMessageHandler(): MessageHandler {
    return MessageHandler { message: Message<*> ->
        try {
            // Extract message information FIRST (before processing)
            val topic = message.headers[MqttHeaders.RECEIVED_TOPIC] as? String ?: ""
            val payload = message.payload as String
            val qos = message.headers[MqttHeaders.RECEIVED_QOS] as? Int

            // Logging for debugging
            logger.debug("MQTT message received - Topic: {}, QoS: {}, Payload: {}",
                topic, qos, payload)

            // Route message to appropriate listener
            when {
                // Legacy topic (backward compatibility)
                topic == "GREENHOUSE" -> {
                    logger.debug("Processing legacy GREENHOUSE topic")
                    greenhouseDataListener.handleGreenhouseData(message)
                }

                // Multi-tenant topic: GREENHOUSE/empresaID
                topic.startsWith("GREENHOUSE/") && topic.split("/").size == 2 -> {
                    val tenantId = topic.substringAfter("GREENHOUSE/")
                    logger.debug("Processing multi-tenant topic for tenant: {}", tenantId)
                    greenhouseDataListener.handleGreenhouseData(message)
                }

                topic.contains("/sensors/") -> sensorDataListener.handleSensorData(message)
                topic.contains("/actuators/status") -> actuatorStatusListener.handleActuatorStatus(message)
                topic.contains("/alerts/") -> logger.info("Alert received on topic: {}", topic)
                else -> logger.warn("Unhandled topic: {}", topic)
            }

        } catch (e: Exception) {
            logger.error("Error processing MQTT message: {}", e.message, e)
            throw e  // Re-throw so error channel handles it
        }
    }
}

mqtt:
  qos:
    default: 0  # Sensor data (high frequency, loss acceptable)

fun handleGreenhouseData(message: Message<*>) {
    val topic = message.headers[MqttHeaders.RECEIVED_TOPIC] as? String ?: return
    val payload = message.payload as String
    
    // Extract tenant ID from topic
    val tenantId = when {
        topic.startsWith("GREENHOUSE/") -> topic.substringAfter("GREENHOUSE/").takeWhile { it != '/' }
        topic == "GREENHOUSE" -> "DEFAULT"
        else -> "UNKNOWN"
    }
    
    // Process the message
    messageProcessor.processGreenhouseData(payload, tenantId)
    
    // ✅ Echo message back to MQTT broker for verification
    try {
        val messageDto = payload.toRealDataDto(
            timestamp = Instant.now(),
            greenhouseId = tenantId
        )
        
        val published = mqttPublishService.publishGreenhouseData(messageDto)
        
        if (published) {
            logger.info("✅ MQTT echo sent successfully - TenantID: {}", tenantId)
        }
    } catch (echoError: Exception) {
        logger.error("❌ Error sending MQTT echo: {}", echoError.message)
    }
}

@Bean
@ServiceActivator(inputChannel = "mqttOutboundChannel")
fun mqttOutbound(): MessageHandler {
    // Generate a unique client ID for outbound adapter
    val clientId = "$clientIdPrefix-outbound-${UUID.randomUUID()}"

    // Create the handler
    val messageHandler = MqttPahoMessageHandler(clientId, mqttClientFactory())

    // Configure the handler
    messageHandler.setAsync(true)  // Async sending for better performance
    messageHandler.setDefaultQos(defaultQos)

    // Note: Topic can be specified per message using MqttHeaders.TOPIC header
    // If not specified, can configure default topic:
    // messageHandler.setDefaultTopic("default/topic")

    return messageHandler
}