/*
 * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 * SPDX-License-Identifier: Apache-2.0.
 */

/**
 *
 * A module containing support for mqtt connection establishment and operations.
 *
 * @packageDocumentation
 * @module mqtt
 * @mergeTarget
 */

import * as internal_mqtt_client from "./mqtt_internal/client";
import * as mqtt5 from "../common/mqtt5";
import * as ws from "./ws"
import * as auth from "./auth";
import * as promise from "../common/promise";
import * as model from "./mqtt_internal/model";
import * as mqtt_shared from "../common/mqtt_shared";
import * as mqtt_shared_browser from "./mqtt_shared_browser";
import * as aws_iot_metrics from "./aws_iot_metrics";

import {Node as TrieNode, Trie, TrieOp} from "./trie";

import {BufferedEventEmitter} from "../common/event";
import {CrtError} from "./error";
import {ClientBootstrap, SocketOptions} from "./io";
import {
    MqttConnectionConnected,
    MqttConnectionDisconnected,
    MqttConnectionResumed,
    MqttRequest,
    MqttSubscribeRequest,
    OnConnectionClosedResult,
    OnConnectionFailedResult,
    OnConnectionSuccessResult,
    OnMessageCallback,
    Payload,
    QoS
} from "../common/mqtt";
import {once} from "events";

export {
    QoS, Payload, MqttRequest, MqttSubscribeRequest, MqttWill, OnMessageCallback, MqttConnectionConnected, MqttConnectionDisconnected,
    MqttConnectionResumed, OnConnectionSuccessResult, OnConnectionFailedResult, OnConnectionClosedResult
} from "../common/mqtt";
export { MqttConnectionConfigBase } from "../common/mqtt_shared";

/**
 * Listener signature for event emitted from an {@link MqttClientConnection} when an error occurs
 *
 * @param error the error that occurred
 *
 * @category MQTT
 */
export type MqttConnectionError = (error: CrtError) => void;

/**
 * Listener signature for event emitted from an {@link MqttClientConnection} when the connection has been
 * interrupted unexpectedly.
 *
 * @param error description of the error that occurred
 *
 * @category MQTT
 */
export type MqttConnectionInterrupted = (error: CrtError) => void;

/**
 * Listener signature for event emitted from an {@link MqttClientConnection} when the connection has been
 * connected successfully.
 *
 * This listener is invoked for every successful connect and every successful reconnect.
 *
 * @param callback_data Data returned containing information about the successful connection.
 *
 * @category MQTT
 */
export type MqttConnectionSuccess = (callback_data: OnConnectionSuccessResult) => void;

/**
 * Listener signature for event emitted from an {@link MqttClientConnection} when the connection has failed
 * to connect.
 *
 * This listener is invoked for every failed connect and every failed reconnect.
 *
 * @param callback_data Data returned containing information about the failed connection.
 *
 * @category MQTT
 */
export type MqttConnectionFailure = (callback_data: OnConnectionFailedResult) => void;

/**
 * Listener signature for event emitted from an {@link MqttClientConnection} when the connection has been
 * disconnected and shutdown successfully.
 *
 * @param callback_data Data returned containing information about the closed/disconnected connection.
 *                      Currently empty, but may contain data in the future.
 *
 * @category MQTT
 */
export type MqttConnectionClosed = (callback_data: OnConnectionClosedResult) => void;

/**
 * @category MQTT
 */
export type WebsocketOptions = ws.WebsocketOptions;

/**
 * @category MQTT
 */
export type AWSCredentials = auth.AWSCredentials;

/**
 * Configuration options for an MQTT connection
 *
 * Extends {@link MqttConnectionConfigBase} with browser-specific options.
 *
 * @category MQTT
 */
export interface MqttConnectionConfig extends mqtt_shared.MqttConnectionConfigBase {
    /** Socket options, ignored in browser */
    socket_options: SocketOptions;

    /** Options for the underlying websocket connection */
    websocket?: WebsocketOptions;

    /** AWS credentials, which will be used to sign the websocket request */
    credentials?: AWSCredentials;

    /** Options for the underlying credentials provider */
    credentials_provider?: auth.CredentialsProvider;
}

/**
 * MQTT client
 *
 * @category MQTT
 */
export class MqttClient {
    constructor(bootstrap?: ClientBootstrap) {

    }

    /**
     * Creates a new {@link MqttClientConnection}
     * @param config Configuration for the connection
     * @returns A new connection
     */
    new_connection(config: MqttConnectionConfig) {
        return new MqttClientConnection(this, config);
    }
}

/**
 * @internal
 */
enum MqttBrowserClientState {
    Connected,
    Stopped
};


/** @internal */
class TopicTrie extends Trie<OnMessageCallback | undefined> {
    constructor() {
        super('/');
    }

    protected find_node(key: string, op: TrieOp) {
        const parts = this.split_key(key);
        let current = this.root;
        let parent = undefined;
        for (const part of parts) {
            let child = current.children.get(part);
            if (!child) {
                child = current.children.get('#');
                if (child) {
                    return child;
                }

                child = current.children.get('+');
            }
            if (!child) {
                if (op == TrieOp.Insert) {
                    current.children.set(part, child = new TrieNode(part));
                }
                else {
                    return undefined;
                }
            }
            parent = current;
            current = child;
        }
        if (parent && op == TrieOp.Delete) {
            parent.children.delete(current.key!);
        }
        return current;
    }
}

interface OneTimeConnectListeners {
    connectionSuccess?: (event: internal_mqtt_client.ConnectionSuccessEvent) => void;
    connectionFailure?: (event: internal_mqtt_client.ConnectionFailureEvent) => void;
}

/**
 * MQTT client connection
 *
 * @category MQTT
 */
export class MqttClientConnection extends BufferedEventEmitter {
    //private connection: mqtt.MqttClient;
    private internalClient: internal_mqtt_client.Client;

    private subscriptions = new TopicTrie();
    private connection_count = 0;

    private connectPromise?: promise.LiftedPromise<boolean> = undefined;
    private isSessionPresent: boolean = false;

    private currentState: MqttBrowserClientState = MqttBrowserClientState.Stopped;
    private desiredState: MqttBrowserClientState = MqttBrowserClientState.Stopped;

    /**
     * @param client The client that owns this connection
     * @param config The configuration for this connection
     */
    constructor(
        readonly client: MqttClient,
        private config: MqttConnectionConfig) {
        super();

        if (!config) {
            throw new CrtError("MqttClientConnection constructor: config not defined");
        }

        let internalConnectOptions : internal_mqtt_client.ConnectOptions = {
            keepAliveIntervalSeconds: this.config.keep_alive ? this.config.keep_alive : 1200,
            resumeSessionPolicy: this.config.clean_session ? internal_mqtt_client.ResumeSessionPolicyType.Never : internal_mqtt_client.ResumeSessionPolicyType.Always,
        };

        if (this.config.client_id) {
            internalConnectOptions.clientId = this.config.client_id;
        }

        if (this.config.disable_metrics) {
            if (this.config.username) {
                internalConnectOptions.username = this.config.username;
            }
        } else {
            internalConnectOptions.username = mqtt_shared_browser.buildFinalUsernameFromMetrics(aws_iot_metrics.create_metrics_mqtt3(this.config), this.config.username);
        }

        if (this.config.password) {
            internalConnectOptions.password = mqtt_shared_browser.normalize_payload_to_buffer(this.config.password);
        }

        if (this.config.will) {
            internalConnectOptions.will = {
                topicName: this.config.will.topic,
                payload: mqtt_shared_browser.normalize_payload_to_buffer(this.config.will.payload),
                qos: this.config.will.qos,
                retain: this.config.will.retain,
            };
        }

        // If the credentials are set but no the credentials_provider
        if (this.config.credentials_provider == undefined &&
            this.config.credentials != undefined) {
            const provider = new auth.StaticCredentialProvider(
                { aws_region: this.config.credentials.aws_region,
                    aws_access_id: this.config.credentials.aws_access_id,
                    aws_secret_key: this.config.credentials.aws_secret_key,
                    aws_sts_token: this.config.credentials.aws_sts_token});
            this.config.credentials_provider = provider;
        }

        let thisConfig = this.config;

        let connectionFactory : () => Promise<ws.WsStream> = function () {
            return new Promise<ws.WsStream>(async (resolve, reject) => {
                if (thisConfig.credentials_provider) {
                    await thisConfig.credentials_provider.refreshCredentials();
                }

                let conn : ws.WsStream = ws.create_websocket_stream(thisConfig);
                conn.on('error', (err) => {
                    reject(err);
                });
                conn.on('connect', () => {
                    resolve(conn);
                });
            });
        }

        let internalConfig : internal_mqtt_client.ClientConfig = {
            protocolVersion: mqtt_shared.ProtocolMode.Mqtt311,
            offlineQueuePolicy: internal_mqtt_client.OfflineQueuePolicy.PreserveQos1PlusPublishes,
            connectOptions: internalConnectOptions,
            pingTimeoutMillis: this.config.ping_timeout ? this.config.ping_timeout : 30 * 1000,
            connectionFactory: connectionFactory,
            // yes this is wrong, but this is how the old client was configured
            // ideally, we'd use the socket options timeout value here
            connectTimeoutMillis: this.config.ping_timeout ? this.config.ping_timeout : 30 * 1000,
            retryJitterMode: mqtt5.RetryJitterType.Default,
            // this was how the previous mqtt-js integration behaved
            resubscribeMode: internal_mqtt_client.ResubscribeModeType.EnabledAlways
        };

        if (config.reconnect_min_sec !== undefined) {
            internalConfig.minReconnectDelayMs = config.reconnect_min_sec * 1000;
        }

        if (config.reconnect_max_sec !== undefined) {
            internalConfig.maxReconnectDelayMs = config.reconnect_max_sec * 1000;
        }

        // we never implemented a reset delay in the browser client, so reset immediately after connecting
        internalConfig.resetConnectionFailureCountMillis = 0;

        this.internalClient = new internal_mqtt_client.Client(internalConfig);

        this.internalClient.on("connectionSuccess", this.on_connection_success);
        this.internalClient.on("connectionFailure", this.on_connection_failure);
        this.internalClient.on("disconnection", this.on_disconnection);
        this.internalClient.on("publishReceived", this.on_publish_received);
        this.internalClient.on("stopped", this.on_stopped);
    }

    /**
     * Emitted when the connection successfully establishes itself for the first time
     *
     * @event
     */
    static CONNECT = 'connect';

    /**
     * Emitted when connection has disconnected successfully.
     *
     * @event
     */
    static DISCONNECT = 'disconnect';

    /**
     * Emitted when an error occurs.  The error will contain the error
     * code and message.
     *
     * @event
     */
    static ERROR = 'error';

    /**
     * Emitted when the connection is dropped unexpectedly. The error will contain the error
     * code and message.  The underlying mqtt implementation will attempt to reconnect.
     *
     * @event
     */
    static INTERRUPT = 'interrupt';

    /**
     * Emitted when the connection reconnects (after an interrupt). Only triggers on connections after the initial one.
     *
     * @event
     */
    static RESUME = 'resume';

    /**
     * Emitted when any MQTT publish message arrives.
     *
     * @event
     */
    static MESSAGE = 'message';

    /**
     * Emitted on every successful connect and reconnect.
     * Will contain a boolean indicating whether the connection resumed a session.
     *
     * @event
     */
    static CONNECTION_SUCCESS = 'connection_success';

    /**
     * Emitted on an unsuccessful connect and reconnect.
     * Will contain an error code indicating the reason for the unsuccessful connection.
     *
     * @event
     */
    static CONNECTION_FAILURE = 'connection_failure';

    /**
     * Emitted when the MQTT connection was disconnected and shutdown successfully.
     *
     * @event
     */
    static CLOSED = 'closed'

    on(event: 'connect', listener: MqttConnectionConnected): this;

    on(event: 'disconnect', listener: MqttConnectionDisconnected): this;

    on(event: 'error', listener: MqttConnectionError): this;

    on(event: 'interrupt', listener: MqttConnectionInterrupted): this;

    on(event: 'connection_success', listener: MqttConnectionSuccess): this;

    on(event: 'connection_failure', listener: MqttConnectionFailure): this;

    on(event: 'closed', listener: MqttConnectionClosed): this;

    on(event: 'resume', listener: MqttConnectionResumed): this;

    on(event: 'message', listener: OnMessageCallback): this;

    on(event: string | symbol, listener: (...args: any[]) => void): this {
        return super.on(event, listener);
    }

    /**
     * Open the actual connection to the server (async).
     * @returns A Promise which completes whether the connection succeeds or fails.
     *          If connection fails, the Promise will reject with an exception.
     *          If connection succeeds, the Promise will return a boolean that is
     *          true for resuming an existing session, or false if the session is new
     */
    async connect() {
        let client = this;
        let internalClient = this.internalClient;

        let isStarted : boolean = client.desiredState == MqttBrowserClientState.Connected;
        this.desiredState = MqttBrowserClientState.Connected;

        setTimeout(() => { this.uncork() }, 0);

        if (this.currentState == MqttBrowserClientState.Connected && isStarted) {
            return new Promise<boolean>((resolve, reject) => {
                resolve(client.isSessionPresent);
            });
        }

        if (!this.connectPromise) {
            this.connectPromise = promise.newLiftedPromise<boolean>();
            let listeners: OneTimeConnectListeners = {};

            listeners.connectionSuccess = function (event: internal_mqtt_client.ConnectionSuccessEvent) {
                if (client.connectPromise) {
                    client.connectPromise.resolve(event.connack.sessionPresent);
                    client.connectPromise = undefined;
                }
                if (listeners.connectionFailure) {
                    internalClient.removeListener("connectionFailure", listeners.connectionFailure);
                }
            };

            listeners.connectionFailure = function (event: internal_mqtt_client.ConnectionFailureEvent) {
                if (client.connectPromise) {
                    client.connectPromise.reject(event.error);
                    client.connectPromise = undefined;
                }
                if (listeners.connectionSuccess) {
                    internalClient.removeListener("connectionSuccess", listeners.connectionSuccess);
                }
            };

            internalClient.once("connectionSuccess", listeners.connectionSuccess);
            internalClient.once("connectionFailure", listeners.connectionFailure);
        }

        if (!isStarted) {
            internalClient.start();
        }

        return this.connectPromise.promise;
    }

    /**
     * The connection will automatically reconnect. To cease reconnection attempts, call {@link disconnect}.
     * To resume the connection, call {@link connect}.
     * @deprecated
     */
    async reconnect() {
        return this.connect();
    }

    /**
     * Publish message (async).
     * If the device is offline, the PUBLISH packet will be sent once the connection resumes.
     *
     * @param topic Topic name
     * @param payload Contents of message
     * @param qos Quality of Service for delivering this message
     * @param retain If true, the server will store the message and its QoS so that it can be
     *               delivered to future subscribers whose subscriptions match the topic name
     * @returns Promise which returns a {@link MqttRequest} which will contain the packet id of
     *          the PUBLISH packet.
     *
     * * For QoS 0, completes as soon as the packet is sent.
     * * For QoS 1, completes when PUBACK is received.
     * * For QoS 2, completes when PUBCOMP is received.
     */
    async publish(topic: string, payload: Payload, qos: QoS, retain: boolean = false): Promise<MqttRequest> {
        let payload_data = mqtt_shared.normalize_payload(payload);

        let options : internal_mqtt_client.PublishOptions = {};
        if (this.config.protocol_operation_timeout) {
            options.timeoutInMillis = this.config.protocol_operation_timeout;
        }

        let publishResult = await this.internalClient.publish({
            topicName: topic,
            qos: qos,
            payload: payload_data,
            retain: retain
        }, options);

        let id = undefined;
        if (qos == QoS.AtLeastOnce) {
            id = (publishResult.packet as model.PubackPacketInternal).packetId;
        }

        return {packet_id: id};
    }

    /**
     * Subscribe to a topic filter (async).
     * The client sends a SUBSCRIBE packet and the server responds with a SUBACK.
     *
     * subscribe() may be called while the device is offline, though the async
     * operation cannot complete successfully until the connection resumes.
     *
     * Once subscribed, `callback` is invoked each time a message matching
     * the `topic` is received. It is possible for such messages to arrive before
     * the SUBACK is received.
     *
     * @param topic Subscribe to this topic filter, which may include wildcards
     * @param qos Maximum requested QoS that server may use when sending messages to the client.
     *            The server may grant a lower QoS in the SUBACK
     * @param on_message Optional callback invoked when message received.
     * @returns Promise which returns a {@link MqttSubscribeRequest} which will contain the
     *          result of the SUBSCRIBE. The Promise resolves when a SUBACK is returned
     *          from the server or is rejected when an exception occurs.
     */
    async subscribe(topic: string, qos: QoS, on_message?: OnMessageCallback): Promise<MqttSubscribeRequest> {
        if (typeof(topic) !== 'string') {
            return Promise.reject("topic is not a string");
        }
        if (typeof(qos) !== 'number') {
            return Promise.reject("qos is not a number");
        }

        this.subscriptions.insert(topic, on_message);

        let options : internal_mqtt_client.SubscribeOptions = {};
        if (this.config.protocol_operation_timeout) {
            options.timeoutInMillis = this.config.protocol_operation_timeout;
        }

        let suback = await this.internalClient.subscribe({
            subscriptions: [
                {
                    topicFilter: topic,
                    qos: qos
                }
            ]
        }, options);

        return {
            topic: topic,
            /*
             * 128 is not modeled in QoS, either on our side nor mqtt-js's side.
             * We have always passed this 128 to the user and it is not reasonable to extend
             * our output type with 128 since it's also our input type and we don't want anyone
             * to pass 128 to us.
             *
             * The 5 client solves this by making the output type a completely separate enum.
             *
             * By doing this cast, we make the type checker ignore this edge case.
             */
            qos: (suback.reasonCodes[0] as number) as QoS
        }
    }

    /**
     * Unsubscribe from a topic filter (async).
     * The client sends an UNSUBSCRIBE packet, and the server responds with an UNSUBACK.
     * @param topic The topic filter to unsubscribe from. May contain wildcards.
     * @returns Promise which returns a {@link MqttRequest} which will contain the packet id
     *          of the UNSUBSCRIBE packet being acknowledged. Promise is resolved when an
     *          UNSUBACK is received from the server or is rejected when an exception occurs.
     */
    async unsubscribe(topic: string): Promise<MqttRequest> {
        if (typeof(topic) !== 'string') {
            return Promise.reject("topic is not a string");
        }

        this.subscriptions.remove(topic);

        let options : internal_mqtt_client.UnsubscribeOptions = {};
        if (this.config.protocol_operation_timeout) {
            options.timeoutInMillis = this.config.protocol_operation_timeout;
        }

        let unsuback = await this.internalClient.unsubscribe({
            topicFilters: [
                topic
            ]
        }, options);

        return {
            packet_id: (unsuback as model.UnsubackPacketInternal).packetId
        };
    }

    /**
     * Close the connection (async).
     * @returns Promise which completes when the connection is closed.
     */
    async disconnect() {
        const isStarted = this.desiredState == MqttBrowserClientState.Connected;
        this.desiredState = MqttBrowserClientState.Stopped;

        if (!isStarted && this.currentState == MqttBrowserClientState.Stopped) {
            return Promise.resolve();
        }

        let stopped = once(this.internalClient, "stopped");

        if (isStarted) {
            this.internalClient.stop();
        }

        return stopped;
    }

    /**
     * Queries whether the client is currently connected
     *
     * @returns whether the client is currently connected
     */
    is_connected() : boolean {
        return this.currentState == MqttBrowserClientState.Connected;
    }

    private on_connection_success = (event: internal_mqtt_client.ConnectionSuccessEvent) => {
        this.on_online(event.connack.sessionPresent);
    }

    private on_online = (session_present: boolean) => {
        this.isSessionPresent = session_present;
        this.currentState = MqttBrowserClientState.Connected;

        if (++this.connection_count == 1) {
            this.emit('connect', session_present);
        } else {
            /** Reset reconnect times after reconnect succeed. */
            this.emit('resume', 0, session_present);
        }

        // Call connection success every time we connect, whether it is a first connect or a reconnect
        let successCallbackData = { session_present: session_present } as OnConnectionSuccessResult;
        this.emit('connection_success', successCallbackData);
    }

    private on_connection_failure = (event: internal_mqtt_client.ConnectionFailureEvent) => {
        let failureCallbackData = { error: event.error } as OnConnectionFailedResult;
        this.emit('connection_failure', failureCallbackData);
    }

    private on_disconnection = (event : internal_mqtt_client.DisconnectionEvent)=> {
        this.currentState = MqttBrowserClientState.Stopped;
        this.emit('interrupt', -1);
    }

    private on_stopped = (event : internal_mqtt_client.StoppedEvent) => {
        this.emit("closed");
        this.emit("disconnect");
    }

    private on_publish_received = (event : internal_mqtt_client.PublishReceivedEvent) => {

        let packet = event.publish as model.PublishPacketInternal;
        let topic = packet.topicName;
        let payload = packet.payload;
        let retain = packet.retain ?? false;

        const array_buffer = payload ? model.payloadToArrayBuffer(payload) : new ArrayBuffer(0);

        const callback = this.subscriptions.find(topic);
        if (callback) {
            callback(topic, array_buffer, packet.duplicate, packet.qos, retain);
        }
        this.emit('message', topic, array_buffer, packet.duplicate, packet.qos, retain);
    }
}