import './ext/array.js';
import { PromiseManager, AbortError, createPromise } from "./async/promise.js";
import { UrlFactory } from './medium/urlfactory.js';
import * as Protocol from './medium/protocol.js';
import * as Nchan from "./medium/nchan.js";
import ProcessingQueue from "./async/queue.js";
import { defineConstants } from "./ext/class.js";
import { defineEvents, ChangeEvent } from "./async/events.js";

const ConnectionState = {
    WAITING: 0,
    CONNECTING: 1,
    CONFIGURING: 2,
    CONNECTED: 3,
    CLOSING: 4,
    CLOSED: 5
}

export class WebSocketError extends Error {
    constructor(message, code) {
        super(message);
        this.code = code;
    }
}


export class Pinger {

    constructor(interval=30) {
        this.interval = interval;
        this.connections = new Set();
        this._timerId = 0;
        this._last = 0;
    }

    get last() { return this._last; }

    get active() { return !!this._timerId; }

    get message() {
        return {
            type: 'ctl',
            name: 'ping',
            timestamp: Date.now()
        };
    }

    start() {
        clearInterval(this._timerId);
        this.interval = Math.max(0, this.interval|0);
        this._timerId = this.interval && setInterval(() => {
            const msg = JSON.stringify(this.message);
            for (const socket of this.connections) {
                if (socket.readyState === WebSocket.OPEN)
                    socket.send(msg);
                else if (socket.readyState !== WebSocket.CONNECTING)
                    this.sockets.delete(socket);
            }
        }, this.interval * 1000);
    }

    stop() {
        clearInterval(this._timerId);
        this._timerId = 0;
    }
}


export class IncrementalWait {

    constructor(options) {
        const {
            max=300000,
            stepFactor=1.5,
            minStep=1000,
            maxStep=3000
        } = { ...options };
        this.currentInterval = 0;
        this.options = {
            max,
            factor: stepFactor,
            min: minStep,
            delta: maxStep - minStep
        };
        this.timer = 0;
    }

    computeNextInterval() {
        return Math.min(
            this.options.max,
            this.currentInterval * this.options.factor +
                this.options.min +
                this.options.delta * Math.random()
        );
    }

    reset() {
        this.tries = 0;
        this.currentInterval = 0;
    }

    async wait() {
        const waitPromise = new Promise(res => this.timer = setTimeout(res, this.currentInterval));
        this.currentInterval = this.computeNextInterval();
        await waitPromise;
    }

    cancel() {
        clearInterval(this.timer);
    }

}

const equalSets = (list1, list2) => {
    const s1 = new Set(list1);
    const s2 = new Set(list2);
    return s1.size === s2.size && s1.isSubsetOf(s2);
}

const nullLogger = {
    log() {},
    info() {},
    warning() {},
    error() {}
};

const OwnChannels = Symbol('all channels');

const implementsUrlFactory = obj =>
    ['restEndpoint', 'publisherEndpoint', 'messengerEndpoint'].every(
        prop => typeof obj[prop] === 'function'
    );

    
export default class Messenger extends EventTarget {

    /**
     * @param {UrlFactory} urlFactory
     * @param {*} options 
     */
    constructor(urlFactory, options) {

        if (!implementsUrlFactory(urlFactory)) {
            throw new TypeError('the first agument must implement a UrlFactory interface');
        }

        super();
        const self = this;

        // pub, sub, url, reconnect
        this.options = {
            waitOnClose: 500,
            autoReconnect: true,
            pingInterval: 0,
            logger: nullLogger,
            ...options
        };

        this.urlFactory = urlFactory;

        /**
         * A set of access tokens
         */
        this.tokens = (function(self) {

            const collection = new Set();
            let merged = null;

            return {
                add(...tokens) {
                    merged = null;
                    tokens.forEach(token => collection.add(token));
                },
                clear() {
                    merged = null;
                    collection.clear();
                },
                delete(...tokens) {
                    merged = null;
                    tokens.forEach(token => collection.delete(token));
                },
                has(token) {
                    return collection.has(token);
                },
                get size() {
                    return collection.size;
                },
                async merged({ signal } = {}) {
                    if (!merged) {
                        if (this.size < 2) {
                            merged = collection.values().next().value;
                        } else {
                            const url = self.urlFactory.restEndpoint('token/merge');
                            const response = await fetch(url, {
                                method: 'POST',
                                signal,
                                body: JSON.stringify({ tokens: [...collection] })
                            }).then(data => data.json());
                            if (response.success) {
                                merged = response.data;
                            } else {
                                throw Error(`Failed to generate an access token: ${response.reason}`);
                            }
                        }
                    }
                    return merged;
                }
            };

        }(this));

        /**
         * Currently connected channels
         */
        this._channels = {
            sub: null,
            pub: null
        };

        /**
         * A public access point to channels
         */
        this.channels = {
            get pub() { return self._channels.pub || []; },
            get sub() { return self._channels.sub || []; }
        };

        this._clientId = null;

        /**
         * The wrapped WebSocket object
         * @type {WebSocket}
         * @private
         */
        this._backend = null;
        
        /**
         * A collection of promises: connect, close, custom promises
         * @type {PromiseManager}
         * @private
         */
        this._promises = new PromiseManager();

        /**
         * An internal list of messages to be sent
         * @type {ProcessingQueue}
         * @private
         */
        this._queue = new ProcessingQueue(msg => { this._backend.send(msg); return true; });
        this._queue.pause();

        /** 
         * The code of the last close event or 0 when none
         * @type {number}
         */
        this.lastExitCode = 0;

        /**
         * The current state of the connection. Exposed via .state getter.
         */
        this._state = ConnectionState.CLOSED;

        this.reconnectController = this.options.autoReconnect ? new IncrementalWait() : null;

        this.pingSender = new Pinger(this.options.pingInterval|0);

        /** @type {Map<string, Set>} */
        this._subscribers = new Map();
        this._subscribers.set(OwnChannels, new Set());

        this._requests = new Map();

        /**
         * A list for received messages that were not yet processed
         */
        this._received = [];

        /**
         * Messages with earlier timestamps are ignored.
         * @type {number}
         */
        this.lastMessageTimestamp = 0;

        /** @type {number} */
        this.lastRequestId = 0;
    }

    get logger() {
        return this.options.logger;
    }

    get clientId() {
        return this._clientId;
    }

    get ownSubChannel() {
        return this._channels.sub[2] || this._channels.sub[0];
    }

    get state() {
        return this._state;
    }

    set state(value) {
        if (this._state === value) return;
        const previous = this._state;
        this._state = value;
        this._promises.resolve('state', value);
        this.dispatchEvent(new ChangeEvent('statechange', { previous, current: value }));
    }

    get desiredState() {
        return this._state < ConnectionState.CLOSING
            ? ConnectionState.CONNECTED
            : ConnectionState.CLOSED;
    }

    get isReady() {
        return this._state === ConnectionState.CONNECTED;
    }

    get reconnectionTime() {
        return this.reconnectController ? this.reconnectController.currentInterval : 0;
    }

    async waitForConnectionState(desiredState, options) {
        if (desiredState === this._state) return;
        await createPromise((res, rej, tearDown) => this.addEventListener(
            'statechange',
            () => (this._state === desiredState) && res(),
            { signal: tearDown }
        ), options);
    }

    async connected(options) {
        await Promise.race([
            this.waitForConnectionState(ConnectionState.CONFIGURING, options),
            this.waitForConnectionState(ConnectionState.CONNECTED, options)
        ]);
    }

    async opened(options) {
        await this.waitForConnectionState(ConnectionState.CONNECTED, options);
    }

    async closed(options) {
        await this.waitForConnectionState(ConnectionState.CLOSED, options);
    }

    async open() {
        if (this.state === ConnectionState.CONNECTED) return;
        const openedPromise = this._promises.create('opened');
        switch (this.state) {
            case ConnectionState.CLOSING:
                this._backend.onerror = null;
                this._backend.onclose = null;
                this._promises.reject('closed', new AbortError('Requested to reconnect'));
                // fallthrough

            case ConnectionState.WAITING:
                this.reconnectController && this.reconnectController.cancel();
                // fallthrough

            case ConnectionState.CLOSED:
                this.state = ConnectionState.CONNECTING;
                // Token is obtained asynchronously: the state might have changed
                // in a meantime. We control it with an abort signal
                const abortController = new AbortController();
                openedPromise.then(
                    () => abortController.abort(),
                    () => abortController.abort()
                );
                try {
                    const token = await this.tokens.merged({ signal: abortController.signal });
                    const url = this.urlFactory.messengerEndpoint(token);
                    this._backend = new WebSocket(url, 'ws+meta.nchan');
                    this._backend.onopen = this.handleSocketOpen.bind(this);
                    this._backend.onclose = this.handleSocketClose.bind(this);
                    this._backend.onerror = this.handleSocketError.bind(this);
                    this._backend.onmessage = this.handleSocketMessage.bind(this);
                } catch (error) {
                    // Forward only fetch related errors - do nothing when aborted
                    // due to the original promise being settled.
                    if (!abortController.signal.aborted) {
                        throw error;
                    }
                }
        }
        await openedPromise;
    }

    async close(discardUnsent=false)
    {
        switch (this.state)
        {
            case ConnectionState.CONNECTING:     // connecting - closing triggers an error, so be careful!
            case ConnectionState.CONFIGURING:
                if (!this._backend) {
                    this._promises.reject('opened', new AbortError("Requested to disconnect"));
                    this.state = ConnectionState.CLOSED;
                    this.lastExitCode = 1000;
                } else {
                    this._backend.onerror = null;
                    this._backend.onopen = null;
                    this._promises.reject('opened', new AbortError("Requested to disconnect"));
                    this.state = ConnectionState.CLOSING;
                    this._backend.close(1000);
                    await this._promises.create('closed');
                }
                break;

            case ConnectionState.CONNECTED:
                this.state = ConnectionState.CLOSING;
                if ((this._backend.bufferedAmount || !this._queue.empty) && !discardUnsent) {
                    this.logger.warning('waiting for queued data to be sent');
                    setTimeout(this.close.bind(this, false), this.options.waitOnClose);
                } else {
                    this.logger.info('closing connection');
                    this._backend.close(1000);    // normal closure
                }
                // fallthrough

            case ConnectionState.CLOSING:
                await this._promises.create('closed');
                break;

            case ConnectionState.WAITING:           // attempting to reconnect
                this.reconnectController && this.reconnectController.cancel();
                this._promises.reject('opened', new AbortError("Requested to disconnect"));
                this.state = ConnectionState.CLOSED;
                break;
        }
        return this.lastExitCode;
    }

    /**
     * Initiates the web socket connection and processes the message queue.
     * 
     * It updates the connection state to CONNECTED, adds the socket to
     * a ping sender if any, resolves the promise labeled by CONNECTED
     * and fires the connect event. The message queue is processed after.
     * 
     * @param {Event} evt
     */
    handleSocketOpen(evt) {
        this.logger.info(`[WebSocket]: onconnected(${this._backend.url})`);
        this.state = ConnectionState.CONFIGURING;
        this.reconnectController && this.reconnectController.reset();
        this.pingSender.connections.add(this._backend);
        // wait until fully initialized before dispatching the event
        this._promises.create('opened').then(
            () => this.dispatchEvent(new evt.constructor(evt.type, evt)),
            () => false
        );
        this._queue.resume();
        this.pingSender.start();
    }

    /**
     * Updates the connection state, removes the socket from a ping sender
     * if any, resolves or rejects the last promise and fires the disconnect
     * event.
     * 
     * The method is invoked in the following situations:
     *   CONNECTING:    the attempt to connect the web socket has failed
     *   CONNECTED:     the connection as been closed due to an error
     *   DISCONNECTING: the connection has been closed by calling disconnect()
     * 
     * @param {CloseEvent} evt    The event object passed to the handler of
     *                            the onclose event
     */
    async handleSocketClose(evt)
    {
        this.logger.info(`[WebSocket]: onclose(${this._backend.url}):`, evt.code);
        this._queue.pause();
        this.pingSender.stop();
        this.pingSender.connections.delete(this._backend);
        this.lastExitCode = evt.code;
        const previousState = this.state;
        const desiredState = this.desiredState;
        this.state = ConnectionState.CLOSED;
        switch (previousState) {
            case ConnectionState.CONNECTING:
                this._promises.reject('opened', new WebSocketError('Failed to open connection', evt.code));
                break;
            case ConnectionState.CLOSING:
                this._promises.resolve('closed', evt.code);
                break;
        }
        // Closing codes
        //   1000 normal closure    Don't reconnect
        //   1001 going away        Don't reconnect when leaving the page (?)
        //   1002 procole error
        //   1003 unsupported data
        //   1005 no status received                    Don't reconnect
        //   1006 abnormal closure (no close frame)     Don't reconnect ?
        //   1007 invalid frame
        //   1008 policy violation
        //   1009 message too big
        //   1010 missing extension (client terminates)     Don't reconnect
        //   1011 internal error (server error)
        //   1012 service restart
        //   1013 try again later (server overloaded)
        //   1014 bad gateway
        //   1015 TLS handshake
        if (desiredState === ConnectionState.CONNECTED && this.reconnectController && ![1000,1001,1005,1010].includes(evt.code)) {
            try {
                this.dispatchEvent(new evt.constructor(evt.type, evt));
                this.state = ConnectionState.WAITING;
                await this.reconnectController.wait();
                // open() returns a promise. Use catch to prevent leaked exceptions
                this.open().catch(() => {});
            } catch (error) {
                // The reconnecting process has been interrupted
                this.logger.error('while reconnecting', error);
                this._promises.reject('open', error);
                this.state = ConnectionState.CLOSED;
            }
        } else {
            this.dispatchEvent(new evt.constructor(evt.type, evt));
        }
    }

    /**
     * A handler for the onerror event.
     * 
     * @param {ErrorEvent} evt    The event object passed to the handler
     *                            of the onerror event
     */
    handleSocketError(evt)
    {
        this.logger.error(evt.message, evt);
        this.dispatchEvent(new evt.constructor(evt.type, evt));
    }

    getSubscribers(channel) {
        const direct = this._subscribers.get(channel) || new Set();
        const all = this._subscribers.get(OwnChannels)|| new Set();
        return new Set([...direct, ...all]);
    }

    /**
     * A handler for the onmessage event.
     * @param {MessageEvent} evt     The event object passed to the handler
     *                               of the onmessage event.
     */
    handleSocketMessage(evt) {
        this.logger.info(`[${this.clientId}] message\n`, evt.data);
        try {
            const parsed = Nchan.parseWebSocketMessage(evt.data);
            parsed.payload = JSON.parse(parsed.payload);

            if (parsed.timestamp < this.lastMessageTimestamp) {
                return;
            } else if (parsed.payload.type === 'ctl') {
                this.processControllMessage(parsed);
            } else if (this._channels.sub) {
                this.processMessage(parsed)
            } else {
                this._received.push(parsed)
            }
            this.lastMessageTimestamp = parsed.timestamp;

        } catch (err) {
            this.logger.error(`[${this.clientId}] message error`, err);
        }
    }

    processControllMessage(message) {
        const { name, payload } = message.payload;
        switch (name) {
            case 'hello':
                this._clientId = payload.clientId;
                this._channels.pub = payload.pub || [];
                this._channels.sub = payload.sub || [];
                this.logger.debug(`[${this.clientId}] connected to group ${payload.group}.\nListens to ${this._channels.sub.join(',')}.\nPublishes to ${this._channels.pub.join(',')}.`);
                if (this.state !== ConnectionState.CONNECTED) {
                    this.logger.info(`[WebSocket]: onopen(${this._backend.url})`);
                    this.state = ConnectionState.CONNECTED;
                    this._promises.resolve('opened');
                }
                // Process queue messages
                const received = [...this._received];
                this._received.length = 0;
                received.forEach(this.processMessage.bind(this));
                break;
        }
    }

    processMessage({ payload, ...message }) {
        
        const { type, name, reqId, reply, payload: content } = payload;
        message.channel = this._channels.sub[message.channel];

        switch (type) {
            case 'message':
                const msgSubscribers = this.getSubscribers(message.channel);
                if (msgSubscribers.size) {
                    Object.assign(message, { name });
                    const msg = new Protocol.Message(content, message);
                    for (const subscriber of msgSubscribers) {
                        if (typeof subscriber.handleMessage === 'function')
                            subscriber.handleMessage(msg);
                    }    
                }
                break;
            case 'request':
                // Make sure that the request is not repeated
                const reqSubscribers = this.getSubscribers(message.channel);
                if (!this._requests.has(reqId) && reqSubscribers.size) {
                    Object.assign(message, { name, reqId, reply });
                    const request = new Protocol.Request(content, message, this);
                    request.ondone = () => this._requests.delete(reqId);
                    this._requests.set(reqId, request);
                    for (const subscriber of reqSubscribers) {
                        if (typeof subscriber.handleRequest === 'function')
                            subscriber.handleRequest(request);
                    }
                }
                break;
            case 'progress':
                const ctl = this._requests.get(reqId);
                this.logger.info(`[${this.clientId}] progress: `, ctl ? 'requested' : 'not requested');
                ctl && ctl.notify(content);
                break;
            case 'response':
                const req = this._requests.get(reqId);
                this.logger.info(`[${this.clientId}] response: `, req ? 'requested' : 'not requested');
                req && req.close(content);    
                break;
        }
        this.lastMessageTimestamp = message.timestamp;
//        this.dispatchEvent(evt);
    }
        
    subscribe(subscriber, channels) {
        if (channels) {
            for(const channel of channels) {
                this._subscribers.has(channel) || this._subscribers.set(channel, new Set());
                this._subscribers.get(channel).add(subscriber);
            }
        } else {
            this._subscribers.get(OwnChannels).add(subscriber);
        }
    }

    unsubscribe(subscriber, channels) {
        if (channels) {
            channels.forEach(channel => {
                const subscribers = this._subscribers.get(channel);
                subscribers && subscribers.delete(subscriber);
            });
        } else {
            this._subscribers.get(OwnChannels).delete(subscriber);
        }
    }

    /**
     * Sends the message when connected and puts it on an internal queue otherwise.
     * @param {string|object} msg   The message to be sent
     */
    async sendRaw(msg, channels=null) {
        msg = JSON.stringify(msg);
        if (!channels || equalSets(channels, this._channels.pub)) {
            this._queue.put(msg);
        } else {
            const url = this.urlFactory.publisherEndpoint(channels);
            const token = await this.tokens.merged();
            await Nchan.publish(url, msg, { authorization: token }).catch(
                err => {
                    this.logger.error(`Failed to publish a message on ${channels} with token ${token}`, err);
                    throw err;
                }
            );
        }
    }


    async publish(name, payload, { channels=null } = {}) {
        await this.sendRaw({ type: 'message', name, payload }, channels);
    }

    async request(name, payload, { channels=null, signal, timeout } = {}) {
        const reqId = `${this.clientId}:${++this.lastRequestId}`;
        const reply = this.ownSubChannel;

        // check that a response can be obtained!
        if (!reply) {
            throw new Error('at least one subscription channel is required to receive a response');
        }

        // wait for a response!
        const ctl = new Protocol.ResponseController({
            signal, timeout
        }, () => this._requests.delete(reqId));
        this._requests.set(reqId, ctl);

        await this.sendRaw({ type: 'request', reqId, reply, name, payload }, channels);
        return ctl.response;
    }

    async drain(options) {
        await this._queue.waitForEmpty(options);
    }

}

defineConstants(Messenger, ConnectionState)
defineEvents(Messenger, 'open', 'close', 'statechange')
