"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.RuntimeWorkerPool = exports.RuntimeWorker = exports.RuntimeWorkerState = void 0;
const http = require("http");
const uuid = require("uuid");
const types_1 = require("./types");
const events_1 = require("events");
const emulatorLogger_1 = require("./emulatorLogger");
const error_1 = require("../error");
var RuntimeWorkerState;
(function (RuntimeWorkerState) {
    RuntimeWorkerState["CREATED"] = "CREATED";
    RuntimeWorkerState["IDLE"] = "IDLE";
    RuntimeWorkerState["BUSY"] = "BUSY";
    RuntimeWorkerState["FINISHING"] = "FINISHING";
    RuntimeWorkerState["FINISHED"] = "FINISHED";
})(RuntimeWorkerState = exports.RuntimeWorkerState || (exports.RuntimeWorkerState = {}));
const FREE_WORKER_KEY = "~free~";
class RuntimeWorker {
    constructor(triggerId, runtime, extensionLogInfo, timeoutSeconds) {
        this.runtime = runtime;
        this.extensionLogInfo = extensionLogInfo;
        this.timeoutSeconds = timeoutSeconds;
        this.stateEvents = new events_1.EventEmitter();
        this.logListeners = [];
        this._state = RuntimeWorkerState.CREATED;
        this.id = uuid.v4();
        this.triggerKey = triggerId || FREE_WORKER_KEY;
        this.runtime = runtime;
        const childProc = this.runtime.process;
        let msgBuffer = "";
        childProc.on("message", (msg) => {
            msgBuffer = this.processStream(msg, msgBuffer);
        });
        let stdBuffer = "";
        if (childProc.stdout) {
            childProc.stdout.on("data", (data) => {
                stdBuffer = this.processStream(data, stdBuffer);
            });
        }
        if (childProc.stderr) {
            childProc.stderr.on("data", (data) => {
                stdBuffer = this.processStream(data, stdBuffer);
            });
        }
        this.logger = triggerId
            ? emulatorLogger_1.EmulatorLogger.forFunction(triggerId, extensionLogInfo)
            : emulatorLogger_1.EmulatorLogger.forEmulator(types_1.Emulators.FUNCTIONS);
        this.onLogs((log) => {
            this.logger.handleRuntimeLog(log);
        }, true);
        childProc.on("exit", () => {
            this.logDebug("exited");
            this.state = RuntimeWorkerState.FINISHED;
        });
    }
    processStream(s, buf) {
        buf += s.toString();
        const lines = buf.split("\n");
        if (lines.length > 1) {
            lines.slice(0, -1).forEach((line) => {
                const log = types_1.EmulatorLog.fromJSON(line);
                this.runtime.events.emit("log", log);
                if (log.level === "FATAL") {
                    this.runtime.events.emit("log", new types_1.EmulatorLog("SYSTEM", "runtime-status", "killed"));
                    this.runtime.process.kill();
                }
            });
        }
        return lines[lines.length - 1];
    }
    readyForWork() {
        this.state = RuntimeWorkerState.IDLE;
    }
    sendDebugMsg(debug) {
        return new Promise((resolve, reject) => {
            this.runtime.process.send(JSON.stringify(debug), (err) => {
                if (err) {
                    reject(err);
                }
                else {
                    resolve();
                }
            });
        });
    }
    request(req, resp, body, debug) {
        if (this.triggerKey !== FREE_WORKER_KEY) {
            this.logInfo(`Beginning execution of "${this.triggerKey}"`);
        }
        const startHrTime = process.hrtime();
        this.state = RuntimeWorkerState.BUSY;
        const onFinish = () => {
            if (this.triggerKey !== FREE_WORKER_KEY) {
                const elapsedHrTime = process.hrtime(startHrTime);
                this.logInfo(`Finished "${this.triggerKey}" in ${elapsedHrTime[0] * 1000 + elapsedHrTime[1] / 1000000}ms`);
            }
            if (this.state === RuntimeWorkerState.BUSY) {
                this.state = RuntimeWorkerState.IDLE;
            }
            else if (this.state === RuntimeWorkerState.FINISHING) {
                this.logDebug(`IDLE --> FINISHING`);
                this.runtime.process.kill();
            }
        };
        return new Promise((resolve) => {
            const reqOpts = Object.assign(Object.assign({}, this.runtime.conn.httpReqOpts()), { method: req.method, path: req.path, headers: req.headers });
            if (this.timeoutSeconds) {
                reqOpts.timeout = this.timeoutSeconds * 1000;
            }
            const proxy = http.request(reqOpts, (_resp) => {
                resp.writeHead(_resp.statusCode || 200, _resp.headers);
                let finished = false;
                const finishReq = (event) => {
                    this.logger.log("DEBUG", `Finishing up request with event=${event}`);
                    if (!finished) {
                        finished = true;
                        onFinish();
                        resolve();
                    }
                };
                _resp.on("pause", () => finishReq("pause"));
                _resp.on("close", () => finishReq("close"));
                const piped = _resp.pipe(resp);
                piped.on("finish", () => finishReq("finish"));
            });
            if (debug) {
                proxy.setSocketKeepAlive(false);
                proxy.setTimeout(0);
            }
            proxy.on("timeout", () => {
                this.logger.log("ERROR", `Your function timed out after ~${this.timeoutSeconds}s. To configure this timeout, see
      https://firebase.google.com/docs/functions/manage-functions#set_timeout_and_memory_allocation.`);
                proxy.destroy();
            });
            proxy.on("error", (err) => {
                this.logger.log("ERROR", `Request to function failed: ${err}`);
                resp.writeHead(500);
                resp.write(JSON.stringify(err));
                resp.end();
                this.runtime.process.kill();
                resolve();
            });
            if (body) {
                proxy.write(body);
            }
            proxy.end();
        });
    }
    get state() {
        return this._state;
    }
    set state(state) {
        if (state === RuntimeWorkerState.IDLE) {
            for (const l of this.logListeners) {
                this.runtime.events.removeListener("log", l);
            }
            this.logListeners = [];
        }
        if (state === RuntimeWorkerState.FINISHED) {
            this.runtime.events.removeAllListeners();
        }
        this.logDebug(state);
        this._state = state;
        this.stateEvents.emit(this._state);
    }
    onLogs(listener, forever = false) {
        if (!forever) {
            this.logListeners.push(listener);
        }
        this.runtime.events.on("log", listener);
    }
    isSocketReady() {
        return new Promise((resolve, reject) => {
            const req = http.request(Object.assign(Object.assign({}, this.runtime.conn.httpReqOpts()), { method: "GET", path: "/__/health" }), () => {
                this.readyForWork();
                resolve();
            });
            req.end();
            req.on("error", (error) => {
                reject(error);
            });
        });
    }
    async waitForSocketReady() {
        const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
        const timeout = new Promise((resolve, reject) => {
            setTimeout(() => {
                reject(new error_1.FirebaseError("Failed to load function."));
            }, 30000);
        });
        while (true) {
            try {
                await Promise.race([this.isSocketReady(), timeout]);
                break;
            }
            catch (err) {
                if (["ECONNREFUSED", "ENOENT"].includes(err === null || err === void 0 ? void 0 : err.code)) {
                    await sleep(100);
                    continue;
                }
                throw err;
            }
        }
    }
    logDebug(msg) {
        this.logger.log("DEBUG", `[worker-${this.triggerKey}-${this.id}]: ${msg}`);
    }
    logInfo(msg) {
        this.logger.logLabeled("BULLET", "functions", msg);
    }
}
exports.RuntimeWorker = RuntimeWorker;
class RuntimeWorkerPool {
    constructor(mode = types_1.FunctionsExecutionMode.AUTO) {
        this.mode = mode;
        this.workers = new Map();
    }
    getKey(triggerId) {
        if (this.mode === types_1.FunctionsExecutionMode.SEQUENTIAL) {
            return "~shared~";
        }
        else {
            return triggerId || "~diagnostic~";
        }
    }
    refresh() {
        for (const arr of this.workers.values()) {
            arr.forEach((w) => {
                if (w.state === RuntimeWorkerState.IDLE) {
                    this.log(`Shutting down IDLE worker (${w.triggerKey})`);
                    w.state = RuntimeWorkerState.FINISHING;
                    w.runtime.process.kill();
                }
                else if (w.state === RuntimeWorkerState.BUSY) {
                    this.log(`Marking BUSY worker to finish (${w.triggerKey})`);
                    w.state = RuntimeWorkerState.FINISHING;
                }
            });
        }
    }
    exit() {
        for (const arr of this.workers.values()) {
            arr.forEach((w) => {
                if (w.state === RuntimeWorkerState.IDLE) {
                    w.runtime.process.kill();
                }
                else {
                    w.runtime.process.kill();
                }
            });
        }
    }
    readyForWork(triggerId) {
        const idleWorker = this.getIdleWorker(triggerId);
        return !!idleWorker;
    }
    async submitRequest(triggerId, req, resp, body, debug) {
        this.log(`submitRequest(triggerId=${triggerId})`);
        const worker = this.getIdleWorker(triggerId);
        if (!worker) {
            throw new error_1.FirebaseError("Internal Error: can't call submitRequest without checking for idle workers");
        }
        if (debug) {
            await worker.sendDebugMsg(debug);
        }
        return worker.request(req, resp, body, !!debug);
    }
    getIdleWorker(triggerId) {
        this.cleanUpWorkers();
        const triggerWorkers = this.getTriggerWorkers(triggerId);
        if (!triggerWorkers.length) {
            this.setTriggerWorkers(triggerId, []);
            return;
        }
        for (const worker of triggerWorkers) {
            if (worker.state === RuntimeWorkerState.IDLE) {
                return worker;
            }
        }
        return;
    }
    addWorker(trigger, runtime, extensionLogInfo) {
        this.log(`addWorker(${this.getKey(trigger === null || trigger === void 0 ? void 0 : trigger.id)})`);
        const disableTimeout = !(trigger === null || trigger === void 0 ? void 0 : trigger.id) || this.mode === types_1.FunctionsExecutionMode.SEQUENTIAL;
        const worker = new RuntimeWorker(trigger === null || trigger === void 0 ? void 0 : trigger.id, runtime, extensionLogInfo, disableTimeout ? undefined : trigger === null || trigger === void 0 ? void 0 : trigger.timeoutSeconds);
        const keyWorkers = this.getTriggerWorkers(trigger === null || trigger === void 0 ? void 0 : trigger.id);
        keyWorkers.push(worker);
        this.setTriggerWorkers(trigger === null || trigger === void 0 ? void 0 : trigger.id, keyWorkers);
        this.log(`Adding worker with key ${worker.triggerKey}, total=${keyWorkers.length}`);
        return worker;
    }
    getTriggerWorkers(triggerId) {
        return this.workers.get(this.getKey(triggerId)) || [];
    }
    setTriggerWorkers(triggerId, workers) {
        this.workers.set(this.getKey(triggerId), workers);
    }
    cleanUpWorkers() {
        for (const [key, keyWorkers] of this.workers.entries()) {
            const notDoneWorkers = keyWorkers.filter((worker) => {
                return worker.state !== RuntimeWorkerState.FINISHED;
            });
            if (notDoneWorkers.length !== keyWorkers.length) {
                this.log(`Cleaned up workers for ${key}: ${keyWorkers.length} --> ${notDoneWorkers.length}`);
            }
            this.setTriggerWorkers(key, notDoneWorkers);
        }
    }
    log(msg) {
        emulatorLogger_1.EmulatorLogger.forEmulator(types_1.Emulators.FUNCTIONS).log("DEBUG", `[worker-pool] ${msg}`);
    }
}
exports.RuntimeWorkerPool = RuntimeWorkerPool;
