mirror of
https://github.com/bourquep/mysa-js-sdk.git
synced 2026-02-04 01:31:05 +00:00
946 lines
34 KiB
TypeScript
946 lines
34 KiB
TypeScript
import { MysaSession } from '@/api/MysaSession';
|
|
import { EventEmitter } from '@/lib/EventEmitter';
|
|
import { parseMqttPayload, serializeMqttPayload } from '@/lib/PayloadParser';
|
|
import { isMsgOutPayload, isMsgTypeOutPayload } from '@/lib/PayloadTypeGuards';
|
|
import { ChangeDeviceState } from '@/types/mqtt/in/ChangeDeviceState';
|
|
import { InMessageType } from '@/types/mqtt/in/InMessageType';
|
|
import { StartPublishingDeviceStatus } from '@/types/mqtt/in/StartPublishingDeviceStatus';
|
|
import { OutMessageType } from '@/types/mqtt/out/OutMessageType';
|
|
import { Devices, DeviceStates, Firmwares } from '@/types/rest';
|
|
import { DescribeThingCommand, IoTClient } from '@aws-sdk/client-iot';
|
|
import { fromCognitoIdentityPool } from '@aws-sdk/credential-providers';
|
|
import {
|
|
AuthenticationDetails,
|
|
CognitoAccessToken,
|
|
CognitoIdToken,
|
|
CognitoRefreshToken,
|
|
CognitoUser,
|
|
CognitoUserPool,
|
|
CognitoUserSession
|
|
} from 'amazon-cognito-identity-js';
|
|
import { io, iot, mqtt } from 'aws-iot-device-sdk-v2';
|
|
import { hash } from 'crypto';
|
|
import dayjs, { Dayjs } from 'dayjs';
|
|
import duration from 'dayjs/plugin/duration.js';
|
|
import { customAlphabet } from 'nanoid';
|
|
import { MqttPublishError, MysaApiError, UnauthenticatedError } from './Errors';
|
|
import { Logger, VoidLogger } from './Logger';
|
|
import { MysaApiClientEventTypes } from './MysaApiClientEventTypes';
|
|
import { MysaApiClientOptions } from './MysaApiClientOptions';
|
|
import { MysaDeviceMode, MysaFanSpeedMode } from './MysaDeviceMode';
|
|
|
|
dayjs.extend(duration);
|
|
|
|
const getRandomClientId = customAlphabet('1234567890abcdefghijklmnopqrstuvwxyz', 8);
|
|
|
|
/** Options for MQTT publish operations. */
|
|
export interface MqttPublishOptions {
|
|
/** Maximum number of publish attempts before failing (default: 5). */
|
|
maxAttempts?: number;
|
|
/** Base delay in milliseconds used for exponential backoff calculation (default: 500). */
|
|
baseDelayMs?: number;
|
|
}
|
|
|
|
const AwsRegion = 'us-east-1';
|
|
const CognitoUserPoolId = 'us-east-1_GUFWfhI7g';
|
|
const CognitoClientId = '19efs8tgqe942atbqmot5m36t3';
|
|
const CognitoIdentityPoolId = 'us-east-1:ebd95d52-9995-45da-b059-56b865a18379';
|
|
const CognitoLoginKey = `cognito-idp.${AwsRegion}.amazonaws.com/${CognitoUserPoolId}`;
|
|
const MqttEndpoint = 'a3q27gia9qg3zy-ats.iot.us-east-1.amazonaws.com';
|
|
const MysaApiBaseUrl = 'https://app-prod.mysa.cloud';
|
|
const RealtimeKeepAliveInterval = dayjs.duration(5, 'minutes');
|
|
|
|
/**
|
|
* Main client for interacting with the Mysa API and real-time device communication.
|
|
*
|
|
* The MysaApiClient provides a comprehensive interface for authenticating with Mysa services, managing device data, and
|
|
* receiving real-time updates from Mysa thermostats and heating devices. It handles both REST API calls for device
|
|
* management and MQTT connections for live status updates and control commands.
|
|
*
|
|
* @example
|
|
*
|
|
* ```typescript
|
|
* const client = new MysaApiClient();
|
|
*
|
|
* await client.login('user@example.com', 'password');
|
|
* const devices = await client.getDevices();
|
|
*
|
|
* client.emitter.on('statusChanged', (status) => {
|
|
* console.log(`Device ${status.deviceId} temperature: ${status.temperature}°C`);
|
|
* });
|
|
*
|
|
* for (const device of Object.entries(devices.DevicesObj)) {
|
|
* await client.startRealtimeUpdates(device[0]);
|
|
* }
|
|
* ```
|
|
*/
|
|
export class MysaApiClient {
|
|
/** The current session object, if any. */
|
|
private _cognitoUserSession?: CognitoUserSession;
|
|
|
|
/** The current user object, if any. */
|
|
private _cognitoUser?: CognitoUser;
|
|
|
|
/** The logger instance used by the client. */
|
|
private _logger: Logger;
|
|
|
|
/** The fetcher function used by the client. */
|
|
private _fetcher: typeof fetch;
|
|
|
|
/** A promise that resolves to the MQTT connection used for real-time updates. */
|
|
private _mqttConnectionPromise?: Promise<mqtt.MqttClientConnection>;
|
|
|
|
/** Stable per-process MQTT client id (prevents collisions between multiple processes). */
|
|
private _mqttClientId?: string;
|
|
|
|
/** Expiration time of the credentials currently in use by the MQTT client. */
|
|
private _mqttCredentialsExpiration?: Dayjs;
|
|
|
|
/** Interrupt timestamps for storm / collision detection. */
|
|
private _mqttInterrupts: number[] = [];
|
|
|
|
/** Whether a forced MQTT reset is currently in progress (guards against re-entrancy). */
|
|
private _mqttResetInProgress = false;
|
|
|
|
/** The device IDs that are currently being updated in real-time, mapped to their respective timeouts. */
|
|
private _realtimeDeviceIds: Map<string, NodeJS.Timeout> = new Map();
|
|
|
|
/** The cached devices object, if any. */
|
|
private _cachedDevices?: Devices;
|
|
|
|
/**
|
|
* Event emitter for client events.
|
|
*
|
|
* @see {@link MysaApiClientEventTypes} for the possible events and their payloads.
|
|
*/
|
|
readonly emitter = new EventEmitter<MysaApiClientEventTypes>();
|
|
|
|
/**
|
|
* Gets the persistable session object.
|
|
*
|
|
* @returns The current persistable session object, if any.
|
|
*/
|
|
get session(): MysaSession | undefined {
|
|
if (!this._cognitoUserSession || !this._cognitoUser) {
|
|
return undefined;
|
|
}
|
|
|
|
return {
|
|
username: this._cognitoUser.getUsername(),
|
|
idToken: this._cognitoUserSession.getIdToken().getJwtToken(),
|
|
accessToken: this._cognitoUserSession.getAccessToken().getJwtToken(),
|
|
refreshToken: this._cognitoUserSession.getRefreshToken().getToken()
|
|
};
|
|
}
|
|
|
|
/**
|
|
* Returns whether the client currently has an active session.
|
|
*
|
|
* @returns True if the client has an active session, false otherwise.
|
|
*/
|
|
get isAuthenticated(): boolean {
|
|
return !!this.session;
|
|
}
|
|
|
|
/**
|
|
* Constructs a new instance of the MysaApiClient.
|
|
*
|
|
* @param session - The persistable session object, if any.
|
|
* @param options - The options for the client.
|
|
*/
|
|
constructor(session?: MysaSession, options?: MysaApiClientOptions) {
|
|
this._logger = options?.logger || new VoidLogger();
|
|
this._fetcher = options?.fetcher || fetch;
|
|
|
|
if (options?.isAwsCrtDebugLoggingEnabled) {
|
|
io.enable_logging(io.LogLevel.DEBUG);
|
|
}
|
|
|
|
if (session) {
|
|
this._cognitoUser = new CognitoUser({
|
|
Username: session.username,
|
|
Pool: new CognitoUserPool({ UserPoolId: CognitoUserPoolId, ClientId: CognitoClientId })
|
|
});
|
|
this._cognitoUserSession = new CognitoUserSession({
|
|
IdToken: new CognitoIdToken({ IdToken: session.idToken }),
|
|
AccessToken: new CognitoAccessToken({ AccessToken: session.accessToken }),
|
|
RefreshToken: new CognitoRefreshToken({ RefreshToken: session.refreshToken })
|
|
});
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Logs in the user with the given email address and password.
|
|
*
|
|
* This method authenticates the user with Mysa's Cognito user pool and establishes a session that can be used for
|
|
* subsequent API calls. Upon successful login, a 'sessionChanged' event is emitted.
|
|
*
|
|
* @example
|
|
*
|
|
* ```typescript
|
|
* try {
|
|
* await client.login('user@example.com', 'password123');
|
|
* console.log('Login successful!');
|
|
* } catch (error) {
|
|
* console.error('Login failed:', error.message);
|
|
* }
|
|
* ```
|
|
*
|
|
* @param emailAddress - The email address of the user.
|
|
* @param password - The password of the user.
|
|
* @throws {@link Error} When authentication fails due to invalid credentials or network issues.
|
|
*/
|
|
async login(emailAddress: string, password: string): Promise<void> {
|
|
this._cognitoUser = undefined;
|
|
this._cognitoUserSession = undefined;
|
|
this._mqttClientId = undefined;
|
|
this._mqttInterrupts = [];
|
|
|
|
this.emitter.emit('sessionChanged', this.session);
|
|
|
|
return new Promise((resolve, reject) => {
|
|
const user = new CognitoUser({
|
|
Username: emailAddress,
|
|
Pool: new CognitoUserPool({ UserPoolId: CognitoUserPoolId, ClientId: CognitoClientId })
|
|
});
|
|
|
|
user.authenticateUser(new AuthenticationDetails({ Username: emailAddress, Password: password }), {
|
|
onSuccess: (session) => {
|
|
this._cognitoUser = user;
|
|
this._cognitoUserSession = session;
|
|
this.emitter.emit('sessionChanged', this.session);
|
|
|
|
resolve();
|
|
},
|
|
onFailure: (err) => {
|
|
reject(err);
|
|
}
|
|
});
|
|
});
|
|
}
|
|
|
|
/**
|
|
* Retrieves the list of devices associated with the user.
|
|
*
|
|
* This method fetches all Mysa devices linked to the authenticated user's account, including device information such
|
|
* as models, locations, and configuration details.
|
|
*
|
|
* @example
|
|
*
|
|
* ```typescript
|
|
* const devices = await client.getDevices();
|
|
* for (const [deviceId, device] of Object.entries(devices.DevicesObj)) {
|
|
* console.log(`Device: ${device.DisplayName} (${device.Model})`);
|
|
* }
|
|
* ```
|
|
*
|
|
* @returns A promise that resolves to the list of devices.
|
|
* @throws {@link MysaApiError} When the API request fails.
|
|
* @throws {@link UnauthenticatedError} When the user is not authenticated.
|
|
*/
|
|
async getDevices(): Promise<Devices> {
|
|
this._logger.debug(`Fetching devices...`);
|
|
|
|
const session = await this._getFreshSession();
|
|
|
|
const response = await this._fetcher(`${MysaApiBaseUrl}/devices`, {
|
|
headers: {
|
|
Authorization: `${session.getIdToken().getJwtToken()}`
|
|
}
|
|
});
|
|
|
|
if (!response.ok) {
|
|
throw new MysaApiError(response);
|
|
}
|
|
|
|
return response.json();
|
|
}
|
|
|
|
/**
|
|
* Retrieves the serial number for a specific device.
|
|
*
|
|
* This method uses AWS IoT's DescribeThing API to fetch the serial number attribute for the specified device. This
|
|
* requires additional AWS IoT permissions and may not be available for all devices.
|
|
*
|
|
* @example
|
|
*
|
|
* ```typescript
|
|
* const serialNumber = await client.getDeviceSerialNumber('device123');
|
|
* if (serialNumber) {
|
|
* console.log(`Device serial: ${serialNumber}`);
|
|
* } else {
|
|
* console.log('Serial number not available');
|
|
* }
|
|
* ```
|
|
*
|
|
* @param deviceId - The ID of the device to get the serial number for.
|
|
* @returns A promise that resolves to the serial number, or undefined if not found.
|
|
* @throws {@link UnauthenticatedError} When the user is not authenticated.
|
|
*/
|
|
async getDeviceSerialNumber(deviceId: string): Promise<string | undefined> {
|
|
this._logger.debug(`Fetching serial number for device ${deviceId}...`);
|
|
|
|
const session = await this._getFreshSession();
|
|
|
|
// Get AWS credentials for IoT client
|
|
const credentialsProvider = fromCognitoIdentityPool({
|
|
clientConfig: {
|
|
region: AwsRegion
|
|
},
|
|
identityPoolId: CognitoIdentityPoolId,
|
|
logins: {
|
|
[CognitoLoginKey]: session.getIdToken().getJwtToken()
|
|
}
|
|
});
|
|
|
|
const credentials = await credentialsProvider();
|
|
const iotClient = new IoTClient({
|
|
region: AwsRegion,
|
|
credentials: {
|
|
accessKeyId: credentials.accessKeyId,
|
|
secretAccessKey: credentials.secretAccessKey,
|
|
sessionToken: credentials.sessionToken
|
|
}
|
|
});
|
|
|
|
try {
|
|
const command = new DescribeThingCommand({ thingName: deviceId });
|
|
const response = await iotClient.send(command);
|
|
return response.attributes?.['Serial'];
|
|
} catch (error) {
|
|
this._logger.warn(`Could not get serial number for device ${deviceId}:`, error);
|
|
return undefined;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Retrieves firmware information for all devices.
|
|
*
|
|
* @returns A promise that resolves to the firmware information for all devices.
|
|
* @throws {@link MysaApiError} When the API request fails.
|
|
* @throws {@link UnauthenticatedError} When the user is not authenticated.
|
|
*/
|
|
async getDeviceFirmwares(): Promise<Firmwares> {
|
|
this._logger.debug(`Fetching device firmwares...`);
|
|
|
|
const session = await this._getFreshSession();
|
|
|
|
const response = await this._fetcher(`${MysaApiBaseUrl}/devices/firmware`, {
|
|
headers: {
|
|
Authorization: `${session.getIdToken().getJwtToken()}`
|
|
}
|
|
});
|
|
|
|
if (!response.ok) {
|
|
throw new MysaApiError(response);
|
|
}
|
|
|
|
return response.json();
|
|
}
|
|
|
|
/**
|
|
* Retrieves the current state information for all devices.
|
|
*
|
|
* @returns A promise that resolves to the current state of all devices.
|
|
* @throws {@link MysaApiError} When the API request fails.
|
|
* @throws {@link UnauthenticatedError} When the user is not authenticated.
|
|
*/
|
|
async getDeviceStates(): Promise<DeviceStates> {
|
|
this._logger.debug(`Fetching device states...`);
|
|
|
|
const session = await this._getFreshSession();
|
|
|
|
const response = await this._fetcher(`${MysaApiBaseUrl}/devices/state`, {
|
|
headers: {
|
|
Authorization: `${session.getIdToken().getJwtToken()}`
|
|
}
|
|
});
|
|
|
|
if (!response.ok) {
|
|
throw new MysaApiError(response);
|
|
}
|
|
|
|
return response.json();
|
|
}
|
|
|
|
/**
|
|
* Sets the state of a specific device by sending commands via MQTT.
|
|
*
|
|
* This method allows you to change the temperature set point and/or operating mode of a Mysa device. The command is
|
|
* sent through the MQTT connection for real-time device control.
|
|
*
|
|
* @example
|
|
*
|
|
* ```typescript
|
|
* // Set temperature to 22°C
|
|
* await client.setDeviceState('device123', 22);
|
|
*
|
|
* // Turn device off
|
|
* await client.setDeviceState('device123', undefined, 'off');
|
|
*
|
|
* // Set temperature and mode
|
|
* await client.setDeviceState('device123', 20, 'heat');
|
|
*
|
|
* // Set fan speed
|
|
* await client.setDeviceState('device123', undefined, undefined, 'auto');
|
|
* ```
|
|
*
|
|
* @param deviceId - The ID of the device to control.
|
|
* @param setPoint - The target temperature set point (optional).
|
|
* @param mode - The operating mode to set (one of MysaDeviceMode values, or undefined to leave unchanged).
|
|
* @param fanSpeed - The fan speed mode to set ('low', 'medium', 'high', 'max', 'auto', or undefined to leave
|
|
* unchanged).
|
|
* @throws {@link UnauthenticatedError} When the user is not authenticated.
|
|
* @throws {@link Error} When MQTT connection or command sending fails.
|
|
*/
|
|
async setDeviceState(deviceId: string, setPoint?: number, mode?: MysaDeviceMode, fanSpeed?: MysaFanSpeedMode) {
|
|
this._logger.debug(`Setting device state for '${deviceId}'`);
|
|
|
|
if (!this._cachedDevices) {
|
|
this._cachedDevices = await this.getDevices();
|
|
}
|
|
|
|
const device = this._cachedDevices.DevicesObj[deviceId];
|
|
|
|
this._logger.debug(`Initializing MQTT connection...`);
|
|
const mqttConnection = await this._getMqttConnection();
|
|
|
|
const now = dayjs();
|
|
|
|
this._logger.debug(`Sending request to set device state for '${deviceId}'...`);
|
|
const modeMap = { off: 1, auto: 2, heat: 3, cool: 4, fan_only: 5, dry: 6 };
|
|
const fanSpeedMap = { auto: 1, low: 3, medium: 5, high: 7, max: 8 };
|
|
|
|
const payload = serializeMqttPayload<ChangeDeviceState>({
|
|
msg: InMessageType.CHANGE_DEVICE_STATE,
|
|
id: now.valueOf(),
|
|
time: now.unix(),
|
|
ver: '1.0',
|
|
src: {
|
|
ref: this.session?.username ?? '',
|
|
type: 100
|
|
},
|
|
dest: {
|
|
ref: deviceId,
|
|
type: 1
|
|
},
|
|
resp: 2,
|
|
body: {
|
|
ver: 1,
|
|
type: device.Model.startsWith('BB-V1')
|
|
? 1
|
|
: device.Model.startsWith('AC-V1')
|
|
? 2
|
|
: device.Model.startsWith('BB-V2')
|
|
? device.Model.endsWith('-L')
|
|
? 5
|
|
: 4
|
|
: 0,
|
|
cmd: [
|
|
{
|
|
tm: -1,
|
|
sp: setPoint,
|
|
md: mode ? modeMap[mode] : undefined,
|
|
fn: fanSpeed ? fanSpeedMap[fanSpeed] : undefined
|
|
}
|
|
]
|
|
}
|
|
});
|
|
|
|
try {
|
|
await this._publishWithRetry(mqttConnection, `/v1/dev/${deviceId}/in`, payload, mqtt.QoS.AtLeastOnce);
|
|
this._logger.debug(`Device state publish succeeded for '${deviceId}'`);
|
|
} catch (error) {
|
|
this._logger.error(`Failed to set device state for '${deviceId}'`, error);
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Starts receiving real-time updates for the specified device.
|
|
*
|
|
* This method establishes an MQTT subscription to receive live status updates from the device, including temperature,
|
|
* humidity, set point changes, and other state information. The client will automatically send keep-alive messages to
|
|
* maintain the connection.
|
|
*
|
|
* @example
|
|
*
|
|
* ```typescript
|
|
* // Start receiving updates and listen for events
|
|
* await client.startRealtimeUpdates('device123');
|
|
*
|
|
* client.emitter.on('statusChanged', (status) => {
|
|
* console.log(`Temperature: ${status.temperature}°C`);
|
|
* });
|
|
* ```
|
|
*
|
|
* @param deviceId - The ID of the device to start receiving updates for.
|
|
* @throws {@link Error} When MQTT connection or subscription fails.
|
|
*/
|
|
async startRealtimeUpdates(deviceId: string) {
|
|
this._logger.info(`Starting real-time updates for device '${deviceId}'`);
|
|
|
|
if (this._realtimeDeviceIds.has(deviceId)) {
|
|
this._logger.debug(`Real-time updates for device '${deviceId}' already started`);
|
|
return;
|
|
}
|
|
|
|
this._logger.debug(`Initializing MQTT connection...`);
|
|
const mqttConnection = await this._getMqttConnection();
|
|
|
|
this._logger.debug(`Subscribing to MQTT topic '/v1/dev/${deviceId}/out'...`);
|
|
await mqttConnection.subscribe(`/v1/dev/${deviceId}/out`, mqtt.QoS.AtLeastOnce, (_, payload) => {
|
|
this._processMqttMessage(payload);
|
|
});
|
|
|
|
this._logger.debug(`Sending request to start publishing device status for '${deviceId}'...`);
|
|
const payload = serializeMqttPayload<StartPublishingDeviceStatus>({
|
|
Device: deviceId,
|
|
MsgType: InMessageType.START_PUBLISHING_DEVICE_STATUS,
|
|
Timestamp: dayjs().unix(),
|
|
Timeout: RealtimeKeepAliveInterval.asSeconds()
|
|
});
|
|
await this._publishWithRetry(mqttConnection, `/v1/dev/${deviceId}/in`, payload, mqtt.QoS.AtLeastOnce);
|
|
|
|
const timer = setInterval(async () => {
|
|
this._logger.debug(`Sending request to keep-alive publishing device status for '${deviceId}'...`);
|
|
|
|
const connection = await this._getMqttConnection();
|
|
const payload = serializeMqttPayload<StartPublishingDeviceStatus>({
|
|
Device: deviceId,
|
|
MsgType: InMessageType.START_PUBLISHING_DEVICE_STATUS,
|
|
Timestamp: dayjs().unix(),
|
|
Timeout: RealtimeKeepAliveInterval.asSeconds()
|
|
});
|
|
await this._publishWithRetry(connection, `/v1/dev/${deviceId}/in`, payload, mqtt.QoS.AtLeastOnce);
|
|
}, RealtimeKeepAliveInterval.subtract(10, 'seconds').asMilliseconds());
|
|
|
|
this._realtimeDeviceIds.set(deviceId, timer);
|
|
}
|
|
|
|
/**
|
|
* Stops receiving real-time updates for the specified device.
|
|
*
|
|
* This method unsubscribes from the MQTT topic for the specified device and clears any associated timers to stop the
|
|
* keep-alive messages.
|
|
*
|
|
* @param deviceId - The ID of the device to stop receiving real-time updates for.
|
|
* @throws {@link Error} When MQTT unsubscription fails.
|
|
*/
|
|
async stopRealtimeUpdates(deviceId: string) {
|
|
this._logger.info(`Stopping real-time updates for device '${deviceId}'`);
|
|
|
|
const timer = this._realtimeDeviceIds.get(deviceId);
|
|
if (!timer) {
|
|
this._logger.warn(`No real-time updates are running for device '${deviceId}'`);
|
|
return;
|
|
}
|
|
|
|
this._logger.debug(`Initializing MQTT connection...`);
|
|
const mqttConnection = await this._getMqttConnection();
|
|
|
|
this._logger.debug(`Unsubscribing to MQTT topic '/v1/dev/${deviceId}/out'...`);
|
|
await mqttConnection.unsubscribe(`/v1/dev/${deviceId}/out`);
|
|
|
|
clearInterval(timer);
|
|
this._realtimeDeviceIds.delete(deviceId);
|
|
}
|
|
|
|
/**
|
|
* Ensures a valid, non-expired session is available.
|
|
*
|
|
* This method checks if the current session is valid and not expired. If the session is expired, it automatically
|
|
* refreshes it using the refresh token.
|
|
*
|
|
* @returns A promise that resolves to a valid CognitoUserSession.
|
|
* @throws {@link UnauthenticatedError} When no session exists or refresh fails.
|
|
*/
|
|
private async _getFreshSession(): Promise<CognitoUserSession> {
|
|
if (!this._cognitoUser || !this._cognitoUserSession) {
|
|
throw new UnauthenticatedError('An attempt was made to access a resource without a valid session.');
|
|
}
|
|
|
|
if (
|
|
this._cognitoUserSession.isValid() &&
|
|
dayjs.unix(this._cognitoUserSession.getIdToken().getExpiration()).isAfter()
|
|
) {
|
|
this._logger.debug('Session is valid, no need to refresh');
|
|
return Promise.resolve(this._cognitoUserSession);
|
|
}
|
|
|
|
this._logger.debug('Session is not valid or expired, refreshing...');
|
|
return new Promise<CognitoUserSession>((resolve, reject) => {
|
|
this._cognitoUser!.refreshSession(this._cognitoUserSession!.getRefreshToken(), (error, session) => {
|
|
if (error) {
|
|
this._logger.error('Failed to refresh session:', error);
|
|
reject(new UnauthenticatedError('Unable to refresh the authentication session.'));
|
|
} else {
|
|
this._logger.debug('Session refreshed successfully');
|
|
this._cognitoUserSession = session;
|
|
this.emitter.emit('sessionChanged', this.session);
|
|
resolve(session);
|
|
}
|
|
});
|
|
});
|
|
}
|
|
|
|
/**
|
|
* Establishes and returns an MQTT connection for real-time communication.
|
|
*
|
|
* This method creates a new MQTT connection if one doesn't exist, using AWS IoT WebSocket connections with Cognito
|
|
* credentials. The connection is cached and reused for subsequent calls.
|
|
*
|
|
* @returns A promise that resolves to an active MQTT connection.
|
|
* @throws {@link Error} When connection establishment fails.
|
|
*/
|
|
private _getMqttConnection(): Promise<mqtt.MqttClientConnection> {
|
|
if (!this._mqttConnectionPromise) {
|
|
this._mqttConnectionPromise = this._createMqttConnection().catch((err) => {
|
|
this._mqttConnectionPromise = undefined;
|
|
throw err;
|
|
});
|
|
}
|
|
|
|
return this._mqttConnectionPromise;
|
|
}
|
|
|
|
/**
|
|
* Determines whether an MQTT-related error is considered transient and worth retrying.
|
|
*
|
|
* Transient errors include timeouts, cancelled operations due to clean sessions, temporary connectivity loss, and
|
|
* other recoverable network issues. Fatal errors (auth, permission, configuration) should not be retried at this
|
|
* layer.
|
|
*
|
|
* @param err - The error object thrown by the underlying MQTT operation.
|
|
* @returns True if the error appears transient and a retry should be attempted; false otherwise.
|
|
*/
|
|
private _isTransientMqttError(err: unknown): boolean {
|
|
if (!err || typeof err !== 'object') {
|
|
return false;
|
|
}
|
|
|
|
const anyErr = err as { error_code?: unknown; error_name?: unknown; error?: unknown; message?: unknown };
|
|
const code = anyErr.error_code || anyErr.error_name || anyErr.error;
|
|
const msg = (anyErr.message || anyErr.error || '').toString();
|
|
|
|
const transientMarkers = [
|
|
'AWS_ERROR_MQTT_TIMEOUT',
|
|
'AWS_ERROR_MQTT_NO_CONNECTION',
|
|
'AWS_ERROR_MQTT_UNEXPECTED_HANGUP',
|
|
'UNEXPECTED_HANGUP',
|
|
'AWS_ERROR_MQTT_CONNECTION_DESTROYED',
|
|
'Time limit between request and response',
|
|
'timeout'
|
|
];
|
|
|
|
return transientMarkers.some((m) => (code && String(code).includes(m)) || msg.includes(m));
|
|
}
|
|
|
|
/**
|
|
* Publishes an MQTT message with exponential backoff retries for transient failures.
|
|
*
|
|
* Retries occur for errors classified by `_isTransientMqttError`. Between attempts the delay grows exponentially with
|
|
* jitter to avoid thundering herds after broker recovery. If the connection is not currently marked as connected, a
|
|
* reconnect is attempted; if that fails, the connection is rebuilt (fresh credentials) before the next retry.
|
|
*
|
|
* On final failure (after maxAttempts) a {@link MqttPublishError} is thrown including the number of attempts and
|
|
* original error for higher-level handling.
|
|
*
|
|
* @remarks
|
|
* Retry options fields:
|
|
*
|
|
* - MaxAttempts: Maximum number of publish attempts before failing (default: 5).
|
|
* - BaseDelayMs: Base delay in milliseconds used for exponential backoff calculation (default: 500).
|
|
*
|
|
* @param connection - The active MQTT client connection used to send the publish.
|
|
* @param topic - The MQTT topic to publish to.
|
|
* @param payload - The serialized payload (binary buffer or Uint8Array).
|
|
* @param qos - The desired MQTT QoS level for the publish.
|
|
* @param opts - Retry options (defaults: maxAttempts=5, baseDelayMs=500).
|
|
* @returns A promise that resolves when the publish succeeds, or rejects with {@link MqttPublishError}.
|
|
*/
|
|
private async _publishWithRetry(
|
|
connection: mqtt.MqttClientConnection,
|
|
topic: string,
|
|
payload: ArrayBuffer | Uint8Array,
|
|
qos: mqtt.QoS,
|
|
opts: MqttPublishOptions = {}
|
|
): Promise<void> {
|
|
const maxAttempts = opts.maxAttempts ?? 5;
|
|
const baseDelayMs = opts.baseDelayMs ?? 500;
|
|
|
|
let attempt = 0;
|
|
|
|
while (true) {
|
|
attempt++;
|
|
try {
|
|
await connection.publish(topic, payload, qos);
|
|
return;
|
|
} catch (err) {
|
|
const isTransient = this._isTransientMqttError(err);
|
|
|
|
if (!isTransient || attempt >= maxAttempts) {
|
|
throw new MqttPublishError(`MQTT publish failed after ${attempt} attempts`, attempt, err);
|
|
}
|
|
|
|
// Apply jitter: delay is randomized between 75% and 125% of the base exponential backoff
|
|
const JITTER_MIN_FACTOR = 0.75;
|
|
const JITTER_RANGE = 0.5;
|
|
const delay = baseDelayMs * Math.pow(2, attempt - 1) * (JITTER_MIN_FACTOR + Math.random() * JITTER_RANGE);
|
|
|
|
this._logger.warn(
|
|
`Transient MQTT publish error on '${topic}' (attempt ${attempt}/${maxAttempts}). Retrying in ${Math.round(
|
|
delay
|
|
)}ms`
|
|
);
|
|
|
|
await new Promise((r) => setTimeout(r, delay));
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Creates a new MQTT connection using AWS IoT WebSocket connections with Cognito credentials.
|
|
*
|
|
* @returns A promise that resolves to an active MQTT connection.
|
|
* @throws {@link Error} When connection establishment fails.
|
|
*/
|
|
private async _createMqttConnection(): Promise<mqtt.MqttClientConnection> {
|
|
const session = await this._getFreshSession();
|
|
const credentialsProvider = fromCognitoIdentityPool({
|
|
clientConfig: {
|
|
region: AwsRegion
|
|
},
|
|
identityPoolId: CognitoIdentityPoolId,
|
|
logins: {
|
|
[CognitoLoginKey]: session.getIdToken().getJwtToken()
|
|
},
|
|
logger: this._logger
|
|
});
|
|
const credentials = await credentialsProvider();
|
|
|
|
if (!credentials.expiration) {
|
|
throw new Error('MQTT credentials do not have an expiration time.');
|
|
}
|
|
|
|
this._mqttCredentialsExpiration = dayjs(credentials.expiration);
|
|
|
|
this._logger.debug(`MQTT credentials expiration: ${this._mqttCredentialsExpiration.format()}`);
|
|
|
|
if (!this._mqttCredentialsExpiration.isAfter(dayjs())) {
|
|
this._mqttCredentialsExpiration = undefined;
|
|
throw new Error('MQTT credentials are already expired.');
|
|
}
|
|
|
|
// Per-process stable client id. Random suffix avoids collisions with other running processes.
|
|
if (!this._mqttClientId) {
|
|
const username = this.session?.username ?? 'anon';
|
|
const usernameHash = hash('sha1', username);
|
|
this._mqttClientId = `mysa-js-sdk-${usernameHash}-${process.pid}-${getRandomClientId()}`;
|
|
}
|
|
|
|
const builder = iot.AwsIotMqttConnectionConfigBuilder.new_with_websockets()
|
|
.with_credentials(AwsRegion, credentials.accessKeyId, credentials.secretAccessKey, credentials.sessionToken)
|
|
.with_endpoint(MqttEndpoint)
|
|
.with_client_id(this._mqttClientId)
|
|
.with_clean_session(false)
|
|
.with_keep_alive_seconds(30)
|
|
.with_ping_timeout_ms(10000)
|
|
.with_protocol_operation_timeout_ms(60000)
|
|
.with_reconnect_min_sec(1)
|
|
.with_reconnect_max_sec(30);
|
|
|
|
const config = builder.build();
|
|
const client = new mqtt.MqttClient();
|
|
const connection = client.new_connection(config);
|
|
|
|
connection.on('connect', () => {
|
|
this._logger.debug(`MQTT connect (clientId=${this._mqttClientId})`);
|
|
});
|
|
|
|
connection.on('connection_success', () => {
|
|
this._logger.debug(`MQTT connection_success (clientId=${this._mqttClientId})`);
|
|
});
|
|
|
|
connection.on('connection_failure', (e) => {
|
|
this._logger.error(`MQTT connection_failure (clientId=${this._mqttClientId})`, e);
|
|
});
|
|
|
|
connection.on('interrupt', async (e) => {
|
|
this._logger.warn(`MQTT interrupt (clientId=${this._mqttClientId})`, e);
|
|
|
|
// Track recent interrupts
|
|
const now = Date.now();
|
|
|
|
// Keep only last 60s
|
|
this._mqttInterrupts = this._mqttInterrupts.filter((t) => now - t < 60000);
|
|
this._mqttInterrupts.push(now);
|
|
|
|
const areCredentialsExpired = !(this._mqttCredentialsExpiration?.isAfter(dayjs()) ?? false);
|
|
|
|
if ((this._mqttInterrupts.length > 5 || areCredentialsExpired) && !this._mqttResetInProgress) {
|
|
this._mqttResetInProgress = true;
|
|
|
|
if (this._mqttInterrupts.length > 5) {
|
|
this._logger.warn(
|
|
`High interrupt rate (${this._mqttInterrupts.length}/60s). Possible clientId collision. Regenerating clientId and resetting connection...`
|
|
);
|
|
} else {
|
|
this._logger.warn(`Credentials expired. Regenerating clientId and resetting connection...`);
|
|
}
|
|
|
|
// Force new client id to escape collision; close current connection
|
|
this._mqttClientId = undefined;
|
|
|
|
this._mqttCredentialsExpiration = undefined;
|
|
|
|
// Clear interrupts
|
|
this._mqttInterrupts = [];
|
|
|
|
// Explicitly clear promise first to prevent reuse while disconnecting
|
|
// (publishers calling _getMqttConnection() will create a new one)
|
|
this._mqttConnectionPromise = undefined;
|
|
|
|
try {
|
|
await connection.disconnect();
|
|
|
|
try {
|
|
this._logger.debug('Old MQTT connection disconnected; establishing new connection...');
|
|
const newConnection = await this._getMqttConnection();
|
|
|
|
for (const deviceId of Array.from(this._realtimeDeviceIds.keys())) {
|
|
const topic = `/v1/dev/${deviceId}/out`;
|
|
this._logger.debug(`Re-subscribing to ${topic}`);
|
|
await newConnection.subscribe(topic, mqtt.QoS.AtLeastOnce, (_topic, payload) => {
|
|
this._processMqttMessage(payload);
|
|
});
|
|
}
|
|
|
|
this._logger.info('MQTT connection rebuilt successfully after interrupt storm or credentials expiration');
|
|
} catch (err) {
|
|
this._logger.error('Failed to re-subscribe after interrupt storm or credentials expiration', err);
|
|
}
|
|
} catch (error) {
|
|
this._logger.error('Error during MQTT reset', error);
|
|
} finally {
|
|
this._mqttResetInProgress = false;
|
|
}
|
|
}
|
|
});
|
|
|
|
connection.on('resume', async (returnCode, sessionPresent) => {
|
|
this._logger.info(
|
|
`MQTT resume returnCode=${returnCode} sessionPresent=${sessionPresent} clientId=${this._mqttClientId}`
|
|
);
|
|
|
|
if (!sessionPresent) {
|
|
this._logger.info('No session present, re-subscribing each device');
|
|
try {
|
|
for (const deviceId of Array.from(this._realtimeDeviceIds.keys())) {
|
|
const topic = `/v1/dev/${deviceId}/out`;
|
|
this._logger.debug(`Re-subscribing to ${topic}`);
|
|
await connection.subscribe(topic, mqtt.QoS.AtLeastOnce, (_topic, payload) => {
|
|
this._processMqttMessage(payload);
|
|
});
|
|
}
|
|
} catch (err) {
|
|
this._logger.error('Failed to re-subscribe after resume', err);
|
|
}
|
|
}
|
|
});
|
|
|
|
connection.on('error', (e) => {
|
|
this._logger.error(`MQTT error (clientId=${this._mqttClientId})`, e);
|
|
});
|
|
|
|
connection.on('closed', () => {
|
|
this._logger.info('MQTT connection closed');
|
|
this._mqttConnectionPromise = undefined;
|
|
this._mqttCredentialsExpiration = undefined;
|
|
});
|
|
|
|
await connection.connect();
|
|
|
|
return connection;
|
|
}
|
|
|
|
/**
|
|
* Processes incoming MQTT messages and emits appropriate events.
|
|
*
|
|
* This method parses MQTT payloads and converts them into typed events that can be listened to via the client's event
|
|
* emitter. It handles both v1 and v2 device message formats and emits events like 'statusChanged', 'setPointChanged',
|
|
* and 'stateChanged'.
|
|
*
|
|
* @param payload - The raw MQTT message payload to process.
|
|
*/
|
|
private _processMqttMessage(payload: ArrayBuffer) {
|
|
try {
|
|
const parsedPayload = parseMqttPayload(payload);
|
|
|
|
this.emitter.emit('rawRealtimeMessageReceived', parsedPayload);
|
|
|
|
if (isMsgTypeOutPayload(parsedPayload)) {
|
|
switch (parsedPayload.MsgType) {
|
|
case OutMessageType.DEVICE_V1_STATUS:
|
|
this.emitter.emit('statusChanged', {
|
|
deviceId: parsedPayload.Device,
|
|
temperature: parsedPayload.MainTemp,
|
|
humidity: parsedPayload.Humidity,
|
|
setPoint: parsedPayload.SetPoint,
|
|
current: parsedPayload.Current
|
|
});
|
|
break;
|
|
|
|
case OutMessageType.DEVICE_SETPOINT_CHANGE:
|
|
this.emitter.emit('setPointChanged', {
|
|
deviceId: parsedPayload.Device,
|
|
newSetPoint: parsedPayload.Next,
|
|
previousSetPoint: parsedPayload.Prev
|
|
});
|
|
break;
|
|
}
|
|
} else if (isMsgOutPayload(parsedPayload)) {
|
|
switch (parsedPayload.msg) {
|
|
case OutMessageType.DEVICE_V2_STATUS:
|
|
this.emitter.emit('statusChanged', {
|
|
deviceId: parsedPayload.src.ref,
|
|
temperature: parsedPayload.body.ambTemp,
|
|
humidity: parsedPayload.body.hum,
|
|
setPoint: parsedPayload.body.stpt,
|
|
dutyCycle: parsedPayload.body.dtyCycle
|
|
});
|
|
break;
|
|
|
|
case OutMessageType.DEVICE_STATE_CHANGE: {
|
|
const modeMap: Record<number, MysaDeviceMode> = {
|
|
1: 'off',
|
|
2: 'auto',
|
|
3: 'heat',
|
|
4: 'cool',
|
|
5: 'fan_only',
|
|
6: 'dry'
|
|
};
|
|
const fanSpeedMap: Record<number, MysaFanSpeedMode> = {
|
|
1: 'auto',
|
|
3: 'low',
|
|
5: 'medium',
|
|
7: 'high',
|
|
8: 'max'
|
|
};
|
|
|
|
this.emitter.emit('stateChanged', {
|
|
deviceId: parsedPayload.src.ref,
|
|
mode: parsedPayload.body.state.md ? modeMap[parsedPayload.body.state.md] : undefined,
|
|
setPoint: parsedPayload.body.state.sp,
|
|
fanSpeed: parsedPayload.body.state.fn !== undefined ? fanSpeedMap[parsedPayload.body.state.fn] : undefined
|
|
});
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
} catch (error) {
|
|
this._logger.error('Error handling MQTT message:', error);
|
|
}
|
|
}
|
|
}
|