Will Bradley
406 lines
12 KiB
JavaScript
406 lines
12 KiB
JavaScript
const mqtt = require('mqtt');
|
|
const config = require('../config/config');
|
|
const logger = require('../utils/logger');
|
|
const { nodeQueries, positionQueries, messageQueries, telemetryQueries } = require('../database/queries');
|
|
|
|
class MeshtasticMQTTClient {
|
|
constructor() {
|
|
this.client = null;
|
|
this.connected = false;
|
|
this.messageCallbacks = [];
|
|
}
|
|
|
|
connect() {
|
|
const { broker, username, password, topic, options } = config.mqtt;
|
|
|
|
logger.info(`Connecting to MQTT broker: ${broker}`);
|
|
|
|
// Create MQTT client
|
|
this.client = mqtt.connect(broker, {
|
|
...options,
|
|
username,
|
|
password
|
|
});
|
|
|
|
// Connection event handlers
|
|
this.client.on('connect', () => {
|
|
this.connected = true;
|
|
logger.info('Connected to MQTT broker');
|
|
|
|
// Subscribe to Meshtastic topics
|
|
this.client.subscribe(topic, (err) => {
|
|
if (err) {
|
|
logger.error('Failed to subscribe to topic:', err);
|
|
} else {
|
|
logger.info(`Subscribed to topic: ${topic}`);
|
|
}
|
|
});
|
|
});
|
|
|
|
this.client.on('error', (error) => {
|
|
logger.error('MQTT connection error:', error);
|
|
this.connected = false;
|
|
});
|
|
|
|
this.client.on('close', () => {
|
|
this.connected = false;
|
|
logger.warn('MQTT connection closed');
|
|
});
|
|
|
|
this.client.on('reconnect', () => {
|
|
logger.info('Reconnecting to MQTT broker...');
|
|
});
|
|
|
|
// Message handler
|
|
this.client.on('message', (topic, message) => {
|
|
this.handleMessage(topic, message);
|
|
});
|
|
|
|
return this;
|
|
}
|
|
|
|
handleMessage(topic, message) {
|
|
try {
|
|
// Parse the topic to extract information
|
|
const topicParts = topic.split('/');
|
|
|
|
// Log raw message for debugging (changed to info for testing)
|
|
logger.info(`Received MQTT message on topic: ${topic} (${message.length} bytes)`);
|
|
|
|
// Try to parse as JSON first (some messages might be JSON)
|
|
let payload;
|
|
try {
|
|
payload = JSON.parse(message.toString());
|
|
this.processJsonMessage(topic, payload);
|
|
} catch (e) {
|
|
// If not JSON, treat as protobuf or binary data
|
|
this.processRawMessage(topic, message);
|
|
}
|
|
|
|
// Notify subscribers
|
|
this.notifyCallbacks(topic, message);
|
|
} catch (error) {
|
|
logger.error('Error handling MQTT message:', error);
|
|
}
|
|
}
|
|
|
|
processJsonMessage(topic, payload) {
|
|
try {
|
|
const { from, to, channel, id, sender } = payload;
|
|
|
|
// Extract node ID (convert to hex string if it's a number, fallback to sender if none
|
|
const fromNode = from ? `!${from.toString(16).padStart(8, '0')}` : sender;
|
|
const toNode = to ? `!${to.toString(16).padStart(8, '0')}` : null;
|
|
|
|
// Debug log for Aether channel
|
|
if (topic.includes('Aether')) {
|
|
logger.info(`Aether message - Type: ${payload.type}, From: ${fromNode}, Channel: ${channel}, Payload: ${JSON.stringify(payload)}`);
|
|
}
|
|
|
|
// Handle different message types based on actual JSON structure
|
|
if ((payload.type === 'sendtext' || payload.type === 'text') && payload.payload) {
|
|
this.handleTextMessage(fromNode, toNode, {
|
|
...payload,
|
|
text: typeof payload.payload === 'string' ? payload.payload : payload.payload.text,
|
|
id: payload.id,
|
|
channel: payload.channel,
|
|
rxSnr: payload.snr,
|
|
rxRssi: payload.rssi
|
|
});
|
|
}
|
|
|
|
if (payload.type === 'position' && payload.payload) {
|
|
const pos = payload.payload;
|
|
this.handlePositionUpdate(fromNode, {
|
|
latitude: pos.latitude_i / 10000000,
|
|
longitude: pos.longitude_i / 10000000,
|
|
altitude: pos.altitude,
|
|
precisionBits: pos.precision_bits
|
|
});
|
|
}
|
|
|
|
if (payload.type === 'nodeinfo' && payload.payload) {
|
|
this.handleNodeInfo(fromNode, {
|
|
shortName: payload.payload.shortname,
|
|
longName: payload.payload.longname,
|
|
hardwareModel: payload.payload.hardware,
|
|
role: payload.payload.role
|
|
});
|
|
}
|
|
|
|
if (payload.type === 'telemetry' && payload.payload) {
|
|
this.handleTelemetry(fromNode, {
|
|
deviceMetrics: {
|
|
batteryLevel: payload.payload.battery_level,
|
|
voltage: payload.payload.voltage,
|
|
channelUtilization: payload.payload.channel_utilization,
|
|
airUtilTx: payload.payload.air_util_tx,
|
|
uptimeSeconds: payload.payload.uptime_seconds
|
|
}
|
|
});
|
|
}
|
|
|
|
if (payload.type === 'neighborinfo' && payload.payload) {
|
|
// Just update last_seen for the node
|
|
nodeQueries.upsertNode.run(
|
|
fromNode,
|
|
null, null, null, null, null,
|
|
null, // last_heard - will use CURRENT_TIMESTAMP
|
|
null, null, null, null, null
|
|
);
|
|
}
|
|
} catch (error) {
|
|
logger.error('Error processing JSON message:', error);
|
|
}
|
|
}
|
|
|
|
processRawMessage(topic, message) {
|
|
// For protobuf messages, we'll need to decode them
|
|
// This is a simplified version - real implementation would use protobuf definitions
|
|
try {
|
|
// Extract information from topic
|
|
const topicParts = topic.split('/');
|
|
|
|
// Basic message logging for debugging
|
|
logger.debug(`Raw message length: ${message.length} bytes`);
|
|
} catch (error) {
|
|
logger.error('Error processing raw message:', error);
|
|
}
|
|
}
|
|
|
|
handleTextMessage(fromNode, toNode, data) {
|
|
try {
|
|
if (!fromNode) return;
|
|
|
|
logger.info(`Text message from ${fromNode}: ${data.text || '(empty)'}`);
|
|
|
|
// Ensure node exists first (upsert)
|
|
nodeQueries.upsertNode.run(
|
|
fromNode,
|
|
null, null, null, null, null,
|
|
null, // last_heard - will use CURRENT_TIMESTAMP via COALESCE
|
|
null, null, null, null, null
|
|
);
|
|
|
|
// Insert message - ensure all values are proper types
|
|
const messageId = data.id ? String(data.id) : null;
|
|
const channel = typeof data.channel === 'number' ? data.channel : 0;
|
|
const text = data.text || '';
|
|
const rxSnr = typeof data.rxSnr === 'number' ? data.rxSnr : null;
|
|
const rxRssi = typeof data.rxRssi === 'number' ? data.rxRssi : null;
|
|
const hopLimit = typeof data.hopLimit === 'number' ? data.hopLimit : null;
|
|
const wantAck = data.wantAck === true ? 1 : 0;
|
|
|
|
messageQueries.insertMessage.run(
|
|
messageId,
|
|
fromNode,
|
|
toNode,
|
|
channel,
|
|
text,
|
|
null, // rx_time - let database use CURRENT_TIMESTAMP for created_at
|
|
rxSnr,
|
|
rxRssi,
|
|
hopLimit,
|
|
wantAck
|
|
);
|
|
} catch (error) {
|
|
logger.error('Error handling text message:', error);
|
|
}
|
|
}
|
|
|
|
handlePositionUpdate(nodeId, data) {
|
|
try {
|
|
if (!nodeId || !data.latitude || !data.longitude) return;
|
|
|
|
logger.info(`Position update from ${nodeId}: ${data.latitude}, ${data.longitude}`);
|
|
|
|
// Update node
|
|
nodeQueries.upsertNode.run(
|
|
nodeId,
|
|
null, // short_name
|
|
null, // long_name
|
|
null, // hardware_model
|
|
null, // role
|
|
null, // firmware_version
|
|
null, // last_heard - will use CURRENT_TIMESTAMP
|
|
null, // battery_level
|
|
null, // voltage
|
|
null, // channel_utilization
|
|
null, // air_util_tx
|
|
null // uptime_seconds
|
|
);
|
|
|
|
// Insert position
|
|
positionQueries.insertPosition.run(
|
|
nodeId,
|
|
data.latitude,
|
|
data.longitude,
|
|
data.altitude || null,
|
|
data.precisionBits || null
|
|
);
|
|
} catch (error) {
|
|
logger.error('Error handling position update:', error);
|
|
}
|
|
}
|
|
|
|
handleNodeInfo(nodeId, data) {
|
|
try {
|
|
if (!nodeId) return;
|
|
|
|
logger.info(`Node info update for ${nodeId}`);
|
|
|
|
// Update node information
|
|
nodeQueries.upsertNode.run(
|
|
nodeId,
|
|
data.shortName || data.user?.shortName || null,
|
|
data.longName || data.user?.longName || null,
|
|
data.hardwareModel || data.user?.hwModel || null,
|
|
data.role || null,
|
|
data.firmwareVersion || null,
|
|
null, // last_heard - will use CURRENT_TIMESTAMP
|
|
null, // battery_level
|
|
null, // voltage
|
|
null, // channel_utilization
|
|
null, // air_util_tx
|
|
null // uptime_seconds
|
|
);
|
|
} catch (error) {
|
|
logger.error('Error handling node info:', error);
|
|
}
|
|
}
|
|
|
|
handleTelemetry(nodeId, data) {
|
|
try {
|
|
if (!nodeId) return;
|
|
|
|
logger.info(`Telemetry update from ${nodeId}`);
|
|
|
|
// Update node with telemetry data
|
|
if (data.deviceMetrics) {
|
|
const metrics = data.deviceMetrics;
|
|
|
|
nodeQueries.upsertNode.run(
|
|
nodeId,
|
|
null, // short_name
|
|
null, // long_name
|
|
null, // hardware_model
|
|
null, // role
|
|
null, // firmware_version
|
|
null, // last_heard - will use CURRENT_TIMESTAMP
|
|
metrics.batteryLevel || null,
|
|
metrics.voltage || null,
|
|
metrics.channelUtilization || null,
|
|
metrics.airUtilTx || null,
|
|
metrics.uptimeSeconds || null
|
|
);
|
|
|
|
// Insert telemetry record
|
|
telemetryQueries.insertTelemetry.run(
|
|
nodeId,
|
|
metrics.batteryLevel || null,
|
|
metrics.voltage || null,
|
|
metrics.channelUtilization || null,
|
|
metrics.airUtilTx || null,
|
|
metrics.uptimeSeconds || null,
|
|
null, // temperature
|
|
null, // relative_humidity
|
|
null // barometric_pressure
|
|
);
|
|
}
|
|
|
|
if (data.environmentMetrics) {
|
|
const env = data.environmentMetrics;
|
|
|
|
telemetryQueries.insertTelemetry.run(
|
|
nodeId,
|
|
null, // battery_level
|
|
null, // voltage
|
|
null, // channel_utilization
|
|
null, // air_util_tx
|
|
null, // uptime_seconds
|
|
env.temperature || null,
|
|
env.relativeHumidity || null,
|
|
env.barometricPressure || null
|
|
);
|
|
}
|
|
} catch (error) {
|
|
logger.error('Error handling telemetry:', error);
|
|
}
|
|
}
|
|
|
|
// Publish a message to MQTT
|
|
publish(topic, message) {
|
|
return new Promise((resolve, reject) => {
|
|
if (!this.connected) {
|
|
reject(new Error('MQTT client not connected'));
|
|
return;
|
|
}
|
|
|
|
this.client.publish(topic, message, (error) => {
|
|
if (error) {
|
|
logger.error('Error publishing message:', error);
|
|
reject(error);
|
|
} else {
|
|
logger.info(`Published message to topic: ${topic}`);
|
|
resolve();
|
|
}
|
|
});
|
|
});
|
|
}
|
|
|
|
// Send a text message
|
|
async sendTextMessage(from, text, channel=0) {
|
|
try {
|
|
const message = JSON.stringify({
|
|
from: from,
|
|
channel: channel,
|
|
type: 'sendtext',
|
|
payload: text
|
|
});
|
|
|
|
// Publish to the appropriate topic
|
|
await this.publish(config.mqtt.pubTopic, message);
|
|
|
|
return true;
|
|
} catch (error) {
|
|
logger.error('Error sending text message:', error);
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
// Register a callback for messages
|
|
onMessage(callback) {
|
|
this.messageCallbacks.push(callback);
|
|
}
|
|
|
|
// Notify all callbacks
|
|
notifyCallbacks(topic, message) {
|
|
this.messageCallbacks.forEach(callback => {
|
|
try {
|
|
callback(topic, message);
|
|
} catch (error) {
|
|
logger.error('Error in message callback:', error);
|
|
}
|
|
});
|
|
}
|
|
|
|
// Get connection status
|
|
isConnected() {
|
|
return this.connected;
|
|
}
|
|
|
|
// Disconnect
|
|
disconnect() {
|
|
if (this.client) {
|
|
this.client.end();
|
|
this.connected = false;
|
|
logger.info('Disconnected from MQTT broker');
|
|
}
|
|
}
|
|
}
|
|
|
|
// Create singleton instance
|
|
const mqttClient = new MeshtasticMQTTClient();
|
|
|
|
module.exports = mqttClient;
|