import {Application, Component, Configurable, Inject} from '@lakutata/core'
import {Logger} from '@lakutata/core/build/plugins/Logger'
import {connect, Socket} from 'socket.io-client'
import {
    CONNECT_PATH, DEFAULT_PORT, FETCH_SERVICES_EVENT, SERVICE_APP_ID,
    SERVICE_APP_NAME, SERVICE_APP_URL, SERVICE_HTTP_ACCESSIBLE_EVENT, SERVICE_PING_EVENT,
    TIMESTAMP_PARAM,
    TRANSPORTS
} from '../../constants/SocketConstant'
import {DisconnectDescription} from 'socket.io-client/build/esm/socket'
import {RequestTimeoutException} from '../../exceptions/RequestTimeoutException'
import {RequestException} from '../../exceptions/RequestException'
import parseUrl from 'parse-url'
import {InvalidRegistryProtocolException} from '../../exceptions/InvalidRegistryProtocolException'
import {ServiceClientInitializationException} from '../../exceptions/ServiceClientInitializationException'
import {Service} from '../../Service'
import {ProviderComponent} from './ProviderComponent'
import {PING_EVENT} from '../../constants/ModuleConstant'
import {ChannelComponent} from './ChannelComponent'
import {IServiceInfoObject} from '../../interfaces/IServiceInfoObject'
import {ConnectionNotReadyException} from '../../exceptions/ConnectionNotReadyException'
import {format} from 'util'
import Fastify, {FastifyInstance} from 'fastify'
import EventEmitter2 from 'eventemitter2'
import axios, {AxiosError} from 'axios'
import {stringify as queryStringStringify} from 'querystring'
import {HttpRequestException} from '../../exceptions/HttpRequestException'
import {HttpRequestEventNotFoundException} from '../../exceptions/HttpRequestEventNotFoundException'
import {IncorrectTokenException} from '../../exceptions/IncorrectTokenException'
import os from 'os'

export class ClientComponent extends Component {

    @Inject(Application)
    protected readonly app: Application

    @Configurable()
    protected readonly service: Service

    @Configurable()
    protected readonly registry: () => string

    @Configurable()
    protected readonly port: () => number

    @Configurable()
    protected readonly token: () => string

    @Configurable()
    protected readonly logger: () => Logger | undefined

    protected readonly acceptProtocols: string[] = ['http', 'https']

    protected readonly registryPattern: RegExp = /^(([^:/?#]+):)?\/\/(([^/?#]+):(.+)@)?([^/?#:]*)(:(\d+))?([^?#]*)(\\?([^#]*))?(#(.*))?/

    protected readonly pingInterval: number = 3000

    protected readonly eventEmitter: EventEmitter2 = new EventEmitter2({
        maxListeners: 10,
        verboseMemoryLeak: true,
        ignoreErrors: true
    })

    protected readonly accessibleTestTimeout: number = 3000

    protected readonly refreshAccessibleURLsInterval: number = 60 * 1000

    public readonly accessibleURLs: Set<string> = new Set()

    protected socket: Socket

    protected httpServer: FastifyInstance

    protected httpServerURL: string

    protected latency: number = 0

    /**
     * 初始化函数
     * @protected
     */
    protected async initialize(): Promise<void> {
        await this.initHttpServer()
        await this.initSocket()
        this.app.once('ready', () => {
            this.refreshAccessibleURLs()
            const refreshAccessibleURLsTimer = setTimeout(() => {
                this.refreshAccessibleURLs()
                    .then(() => refreshAccessibleURLsTimer.refresh())
                    .catch(() => refreshAccessibleURLsTimer.refresh())
            }, this.refreshAccessibleURLsInterval)
        })
    }

    /**
     * http初始化
     * @protected
     */
    protected async initHttpServer(): Promise<void> {
        this.httpServer = await Fastify({
            bodyLimit: 1024 * (1024 * 1024),//1GiB
            logger: false
        })
        this.httpServer.register(import('@fastify/compress'))
        const pathname: string = `/${this.app.Security.generateRandomString(16)}`
        this.httpServer.post(pathname, async (request, reply) => {
            if (typeof request.query === 'object' && typeof request.body === 'object') {
                const {event, token} = request.query as { event: string; token: string }
                if (event) {
                    if (token === this.token()) {
                        if (this.eventEmitter.eventNames().includes(event)) {
                            const payload: any = request.body ? (request.body as any).payload : undefined
                            const result = await this.eventEmitter.emitAsync(event, payload)
                            return {payload: result[0]}
                        }
                        reply.statusCode = 404
                        return
                    }
                    reply.statusCode = 403
                    return
                }
            }
            reply.statusCode = 500
            return
        })
        const urlObject: URL = new URL(await this.httpServer.listen({port: this.port(), host: '0.0.0.0'}))
        urlObject.hostname = this.getLocalIPAddress()
        urlObject.pathname = pathname
        this.httpServerURL = urlObject.toString()
        this.onHttpRequest(SERVICE_HTTP_ACCESSIBLE_EVENT, (accessibleTestData: { nonce: string }) => accessibleTestData)
        return this.app.Logger.info('Service client listening URL:', this.httpServerURL)
    }

    /**
     * 获取本地IP地址
     * @protected
     */
    protected getLocalIPAddress(): string {
        const networkInterfaces = os.networkInterfaces()
        const addresses: string[] = []
        Object.keys(networkInterfaces).forEach(ethernetInterfaceName => {
            networkInterfaces[ethernetInterfaceName]?.forEach(networkInfo => {
                if (networkInfo.internal || networkInfo.family !== 'IPv4') return
                addresses.push(networkInfo.address)
            })
        })
        if (!addresses.length) return '127.0.0.1'
        return addresses[0]
    }

    /**
     * socket初始化
     * @protected
     */
    protected async initSocket(): Promise<void> {
        return new Promise((resolve, reject) => {
            try {
                let portInRegistry: string = ''
                let registryPatternInput: string = this.registry()
                if (!this.registryPattern.test(registryPatternInput)) {
                    registryPatternInput = `registry://${registryPatternInput}`
                }
                if (this.registryPattern.test(registryPatternInput)) {
                    const matches = registryPatternInput.match(this.registryPattern)
                    if (matches) {
                        portInRegistry = matches[8] || ''
                    }
                }
                if (!this.registry()) return reject(this.generateException(ServiceClientInitializationException, 'Service registry address not set'))
                const parsedUrl = parseUrl(this.registry(), true)
                if (!this.acceptProtocols.includes(parsedUrl.protocol)) return reject(this.generateException(InvalidRegistryProtocolException, `Invalid registry protocol [${parsedUrl.protocol}]`))
                parsedUrl.port = parsedUrl.port ? parsedUrl.port : portInRegistry ? portInRegistry : DEFAULT_PORT.toString()
                const urlObject = new URL(parsedUrl.href)
                urlObject.port = parsedUrl.port
                this.socket = connect(urlObject.toString(), {
                    autoConnect: false,//使用当前类内connect方法进行连接
                    reconnection: true,//需要设定为自动重连，否则已绑定socket的各种事件将不容易处理
                    transports: TRANSPORTS,
                    path: CONNECT_PATH,
                    timestampParam: TIMESTAMP_PARAM,
                    timestampRequests: true,
                    extraHeaders: {
                        [SERVICE_APP_ID]: Buffer.from(this.app.getID()).toString('base64'),
                        [SERVICE_APP_NAME]: Buffer.from(this.app.getName()).toString('base64'),
                        [SERVICE_APP_URL]: Buffer.from(this.httpServerURL).toString('base64')
                    },
                    auth: (cb: (data: object) => void) => {
                        cb({
                            token: this.token(),
                            patterns: this.service.Components.get<ProviderComponent>('provider').getPatterns(),
                            events: this.service.Components.get<ChannelComponent>('channel').channelEvents
                        })
                    }
                })
                //定时向注册中心汇报自身状况
                setInterval(() => this.ping(), this.pingInterval)
                this.service.on(PING_EVENT, () => this.ping())
                return resolve()
            } catch (e) {
                return reject(this.generateException(ServiceClientInitializationException, (e as Error).message))
            }
        })
    }

    /**
     * 向注册中心发送ping包
     * @protected
     */
    protected ping(): void {
        try {
            const latencyTestStart: number = Date.now()
            let tasks: number = this.service.Components.get<ProviderComponent>('provider')?.getTasks()
            tasks = tasks ? tasks : 0
            this.socket.volatile.emit(SERVICE_PING_EVENT, {
                events: this.service.Components.get<ChannelComponent>('channel').channelEvents,
                latency: this.latency,
                tasks: tasks
            }, () => {
                this.latency = Date.now() - latencyTestStart
            })
        } catch (e) {
            this.app.Logger.error('Service client ping error:', (e as Error).message)
        }
    }

    /**
     * 连接服务注册中心
     */
    public async connect(): Promise<void> {
        return new Promise((resolve, reject) => {
            try {
                let fulfilled: boolean = false
                this.socket
                    .on('connect', async () => {
                        if (!fulfilled) {
                            fulfilled = true
                            return resolve()
                        } else {
                            this.service.emit(PING_EVENT)
                            return this.logger()?.info('Service client reconnected to registry', this.registry())
                        }
                    })
                    .on('connect_error', (connectError: Error) => this.logger()?.error('Service client cannot connect to', this.registry(), ':', connectError.message))
                    .on('disconnect', (reason: Socket.DisconnectReason, description: DisconnectDescription | undefined) => {
                        const desc: string = (description && !(description instanceof Error)) ? description.description : ''
                        return this.logger()?.warning('Service client disconnected from', this.registry(), ':', reason, desc ? `(${desc})` : '')
                    })
                    .connect()
            } catch (e) {
                return reject(e)
            }
        })
    }

    /**
     * 发送不需要响应的请求数据(socket)
     * @param event
     * @param data
     */
    public socketSend(event: string, data: any): boolean {
        try {
            if (this.socket.connected) {
                this.socket.emit(event, data)
                return true
            }
            return false
        } catch (e) {
            this.app.Logger.error('Send data error:', (e as Error).message)
            return false
        }
    }

    /**
     * 发送请求(http)
     * @param url
     * @param event
     * @param data
     * @param timeout
     */
    public async httpRequest(url: string, event: string, data: any, timeout: number = -1): Promise<any> {
        try {
            const urlObject = new URL(url)
            urlObject.search = queryStringStringify({
                event: event,
                token: this.token()
            })
            const response = await axios.post(urlObject.toString(), {payload: data}, {
                maxRedirects: 16,
                timeout: timeout > -1 ? timeout : undefined
            })
            return response.data?.payload
        } catch (e) {
            const error: AxiosError = (e as AxiosError)
            switch (error.code) {
                case AxiosError.ERR_BAD_REQUEST: {
                    if (error.response!.status === 403) {
                        throw this.generateException(IncorrectTokenException, 'Incorrect token')
                    } else if (error.response!.status === 404) {
                        throw this.generateException(HttpRequestEventNotFoundException, `Event ${event} not found`)
                    }
                    throw this.generateException(HttpRequestException, error.message)
                }
                case AxiosError.ETIMEDOUT: {
                    throw this.generateException(RequestTimeoutException, `Local request timeout after ${timeout}ms, payload: ${format(data)}`)
                }
                default: {
                    throw this.generateException(HttpRequestException, error.message)
                }
            }
        }
    }

    /**
     * 发送请求(socket)
     * @param event
     * @param data
     * @param timeout
     */
    public async socketRequest(event: string, data: any, timeout: number = -1): Promise<any> {
        return new Promise((resolve, reject) => {
            try {
                if (this.socket.connected) {
                    if (timeout > -1) {
                        this.socket.timeout(timeout).emit(event, data, (error: Error, response: any) => error ? reject(this.generateException(RequestTimeoutException, `Local request timeout after ${timeout}ms, payload: ${format(data)}`)) : resolve(response))
                    } else {
                        this.socket.emit(event, data, (response: any) => resolve(response))
                    }
                } else {
                    reject(this.generateException(ConnectionNotReadyException, 'Local connection not ready'))
                }
            } catch (e) {
                reject(this.generateException(RequestException, (e as Error).message))
            }
        })
    }

    /**
     * 响应请求(http)
     * @param event
     * @param callback
     */
    public onHttpRequest(event: string, callback: (data: any) => Promise<any> | any): this {
        this.eventEmitter.on(event, callback)
        return this
    }

    /**
     * 响应请求(socket)
     * @param event
     * @param callback
     */
    public onSocketRequest(event: string, callback: (data: any) => Promise<any> | any): this {
        this.socket.on(event, async (data: any, fn: (response: any) => void) => {
            if (typeof fn === 'function') fn(await callback(data))
        })
        return this
    }

    /**
     * 处理不需要响应的请求(socket)
     * @param event
     * @param callback
     */
    public onSocketMessage(event: string, callback: (data: any) => Promise<void> | void): this {
        this.socket.on(event, (data: any) => {
            return callback(data)
        })
        return this
    }

    /**
     * 从Registry获取集群内的所有服务信息
     * @param appId
     */
    public async fetchServices(appId?: string | RegExp): Promise<IServiceInfoObject> {
        const services: IServiceInfoObject = {}
        const rawJsonStr: string = await this.socketRequest(FETCH_SERVICES_EVENT, null)
        const parsedResponse: { [key: string]: any[] } = JSON.parse(rawJsonStr)
        const appIds: string[] = Object.keys(parsedResponse)
        appIds.forEach(appId => {
            services[appId] = parsedResponse[appId].map(value => {
                return Object.assign({}, value, {
                    events: new Map(Object.entries(value.events))
                })
            })
        })
        if (appId) {
            const regExp: RegExp = typeof appId === 'string' ? new RegExp(appId) : appId
            Object.keys(services).forEach((serviceAppId: string) => {
                if (!regExp.test(serviceAppId)) {
                    delete services[serviceAppId]
                }
            })
        }
        return services
    }

    /**
     * 更新可直接通过http访问的服务URL地址列表
     */
    public async refreshAccessibleURLs(): Promise<void> {
        try {
            const services: IServiceInfoObject = await this.fetchServices()
            const accessibleTestPromises: Promise<string | null>[] = []
            Object.keys(services).forEach(serviceId => {
                services[serviceId].forEach(serviceNodeInfo => {
                    accessibleTestPromises.push(new Promise((resolve) => {
                        const nonce: string = this.app.Security.generateRandomString(16)
                        this.httpRequest(serviceNodeInfo.appURL, SERVICE_HTTP_ACCESSIBLE_EVENT, {nonce: nonce}, this.accessibleTestTimeout)
                            .then(response => resolve(response.nonce === nonce ? serviceNodeInfo.appURL : null))
                            .catch(() => resolve(null))
                    }))
                })
            })
            const accessibleURLs: string[] = (await Promise.all(accessibleTestPromises)).filter(value => value !== null) as string[]
            if (accessibleURLs.length) {
                this.accessibleURLs.forEach((value, value2, set) => {
                    if (!accessibleURLs.includes(value)) set.delete(value)
                })
            } else {
                this.accessibleURLs.clear()
            }
            accessibleURLs.forEach(accessibleURL => {
                if (accessibleURL !== null) this.accessibleURLs.add(accessibleURL)
            })
        } catch (e) {
            this.accessibleURLs.clear()
        }
    }
}