"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.AdminServer = void 0;
const _ = require("lodash");
const express = require("express");
const cors = require("cors");
const corsGate = require("cors-gate");
const bodyParser = require("body-parser");
const Ws = require("ws");
const uuid_1 = require("uuid");
const express_1 = require("graphql-http/lib/use/express");
const graphql_1 = require("graphql");
const graphql_tag_1 = require("graphql-tag");
const schema_1 = require("@graphql-tools/schema");
const subscriptions_transport_ws_1 = require("@httptoolkit/subscriptions-transport-ws");
const stream_1 = require("stream");
const DuplexPair = require("native-duplexpair");
const destroyable_server_1 = require("destroyable-server");
const error_1 = require("../util/error");
const promise_1 = require("../util/promise");
const types_1 = require("../types");
const graphql_utils_1 = require("./graphql-utils");
async function strictOriginMatch(origin, expectedOrigin) {
    if (!origin)
        return false;
    if (typeof expectedOrigin === 'string') {
        return expectedOrigin === origin;
    }
    if (_.isRegExp(expectedOrigin)) {
        return !!origin.match(expectedOrigin);
    }
    if (_.isArray(expectedOrigin)) {
        return _.some(expectedOrigin, (exp) => strictOriginMatch(origin, exp));
    }
    if (_.isFunction(expectedOrigin)) {
        return new Promise((resolve, reject) => {
            expectedOrigin(origin, (error, result) => {
                if (error)
                    reject(error);
                else
                    resolve(strictOriginMatch(origin, result));
            });
        });
    }
    // We don't allow boolean or undefined matches
    return false;
}
class AdminServer {
    constructor(options = {}) {
        this.app = express();
        this.server = null;
        this.eventEmitter = new stream_1.EventEmitter();
        this.sessions = {};
        this.debug = options.debug || false;
        if (this.debug)
            console.log('Admin server started in debug mode');
        this.webSocketKeepAlive = options.webSocketKeepAlive || undefined;
        this.ruleParams = options.ruleParameters || {};
        this.adminPlugins = options.adminPlugins || {};
        if (options.corsOptions?.allowPrivateNetworkAccess) {
            // Allow web pages on non-local URLs (testsite.example.com, not localhost) to
            // send requests to this admin server too. Without this, those requests will
            // fail after rejected preflights in recent Chrome (from ~v102, ish? Unclear).
            // This is combined with the origin restrictions that may be set, so only
            // accepted origins will be allowed to make these requests.
            this.app.use((req, res, next) => {
                if (req.headers["access-control-request-private-network"]) {
                    res.setHeader("access-control-allow-private-network", "true");
                }
                next(null);
            });
        }
        this.app.use(cors(options.corsOptions));
        // If you use strict CORS, and set a specific origin, we'll enforce it:
        this.requiredOrigin = !!options.corsOptions &&
            !!options.corsOptions.strict &&
            !!options.corsOptions.origin &&
            typeof options.corsOptions.origin !== 'boolean' &&
            options.corsOptions.origin;
        if (this.requiredOrigin) {
            this.app.use(corsGate({
                strict: true,
                allowSafe: false,
                origin: '' // No base origin - we accept *no* same-origin requests
            }));
        }
        this.app.use(bodyParser.json({ limit: '50mb' }));
        const defaultPluginStartParams = options.pluginDefaults ?? {};
        this.app.post('/start', async (req, res) => {
            try {
                const rawConfig = req.body;
                // New clients send: "{ plugins: { http: {...}, webrtc: {...} } }" etc. Old clients just send
                // the HTTP options bare with no wrapper, so we wrap them for backward compat.
                const isPluginAwareClient = ('plugins' in rawConfig);
                const providedPluginStartParams = (!isPluginAwareClient
                    ? {
                        http: {
                            options: _.cloneDeep(rawConfig),
                            port: (typeof req.query.port === 'string')
                                ? JSON.parse(req.query.port)
                                : undefined
                        }
                    }
                    : rawConfig.plugins);
                // For each plugin that was specified, we pull default params into their start params.
                const pluginStartParams = _.mapValues((providedPluginStartParams), (params, pluginId) => {
                    return _.merge({}, defaultPluginStartParams[pluginId], params);
                });
                if (this.debug)
                    console.log('Admin server starting mock session with config', pluginStartParams);
                // Backward compat: do an explicit check for HTTP port conflicts
                const httpPort = pluginStartParams.http?.port;
                if (_.isNumber(httpPort) && this.sessions[httpPort] != null) {
                    res.status(409).json({
                        error: `Cannot start: mock server is already running on port ${httpPort}`
                    });
                    return;
                }
                const missingPluginId = Object.keys(pluginStartParams).find(pluginId => !(pluginId in this.adminPlugins));
                if (missingPluginId) {
                    res.status(400).json({
                        error: `Request to mock using unrecognized plugin: ${missingPluginId}`
                    });
                    return;
                }
                const sessionPlugins = _.mapValues(pluginStartParams, (__, pluginId) => {
                    const PluginType = this.adminPlugins[pluginId];
                    return new PluginType();
                });
                const pluginStartResults = await (0, promise_1.objectAllPromise)(_.mapValues(sessionPlugins, (plugin, pluginId) => plugin.start(pluginStartParams[pluginId])));
                // More backward compat: old clients assume that the port is also the management id.
                const sessionId = isPluginAwareClient
                    ? (0, uuid_1.v4)()
                    : sessionPlugins.http.getMockServer().port.toString();
                await this.startSessionManagementAPI(sessionId, sessionPlugins);
                if (isPluginAwareClient) {
                    res.json({
                        id: sessionId,
                        pluginData: _.mapValues(pluginStartResults, (r) => r ?? {} // Always return _something_, even if the plugin returns null/undefined.
                        )
                    });
                }
                else {
                    res.json({
                        id: sessionId,
                        ...(pluginStartResults['http'])
                    });
                }
            }
            catch (e) {
                res.status(500).json({ error: `Failed to start mock session: ${((0, error_1.isErrorLike)(e) && e.message) || e}` });
            }
        });
        this.app.post('/reset', async (req, res) => {
            try {
                await this.resetAdminServer();
                res.json({ success: true });
            }
            catch (e) {
                res.status(500).json({
                    error: ((0, error_1.isErrorLike)(e) && e.message) || 'Unknown error'
                });
            }
        });
        // Dynamically route to mock sessions ourselves, so we can easily add/remove
        // sessions as we see fit later on.
        const sessionRequest = (req, res, next) => {
            const sessionId = req.params.id;
            const sessionRouter = this.sessions[sessionId]?.router;
            if (!sessionRouter) {
                res.status(404).send('Unknown mock session');
                console.error(`Request for unknown mock session with id: ${sessionId}`);
                return;
            }
            sessionRouter(req, res, next);
        };
        this.app.use('/session/:id/', sessionRequest);
        this.app.use('/server/:id/', sessionRequest); // Old URL for backward compat
    }
    async resetAdminServer() {
        if (this.debug)
            console.log('Resetting admin server');
        await Promise.all(Object.values(this.sessions).map(({ stop }) => stop()));
    }
    on(event, listener) {
        this.eventEmitter.on(event, listener);
    }
    async start(listenOptions = types_1.DEFAULT_ADMIN_SERVER_PORT) {
        if (this.server)
            throw new Error('Admin server already running');
        await new Promise((resolve, reject) => {
            this.server = (0, destroyable_server_1.makeDestroyable)(this.app.listen(listenOptions, resolve));
            this.server.on('error', reject);
            this.server.on('upgrade', async (req, socket, head) => {
                const reqOrigin = req.headers['origin'];
                if (this.requiredOrigin && !await strictOriginMatch(reqOrigin, this.requiredOrigin)) {
                    console.warn(`Websocket request from invalid origin: ${req.headers['origin']}`);
                    socket.destroy();
                    return;
                }
                const isSubscriptionRequest = req.url.match(/^\/(?:server|session)\/([\w\d\-]+)\/subscription$/);
                const isStreamRequest = req.url.match(/^\/(?:server|session)\/([\w\d\-]+)\/stream$/);
                const isMatch = isSubscriptionRequest || isStreamRequest;
                if (isMatch) {
                    const sessionId = isMatch[1];
                    let wsServer = isSubscriptionRequest
                        ? this.sessions[sessionId]?.subscriptionServer.server
                        : this.sessions[sessionId]?.streamServer;
                    if (wsServer) {
                        wsServer.handleUpgrade(req, socket, head, (ws) => {
                            wsServer.emit('connection', ws, req);
                        });
                    }
                    else {
                        console.warn(`Websocket request for unrecognized mock session: ${sessionId}`);
                        socket.destroy();
                    }
                }
                else {
                    console.warn(`Unrecognized websocket request for ${req.url}`);
                    socket.destroy();
                }
            });
        });
    }
    async startSessionManagementAPI(sessionId, plugins) {
        const mockSessionRouter = express.Router();
        let running = true;
        const stopSession = async () => {
            if (!running)
                return;
            running = false;
            this.eventEmitter.emit('mock-session-stopping', plugins);
            const session = this.sessions[sessionId];
            delete this.sessions[sessionId];
            await Promise.all(Object.values(plugins).map(plugin => plugin.stop()));
            session.subscriptionServer.close();
            // Close with code 1000 (purpose is complete - no more streaming happening)
            session.streamServer.clients.forEach((client) => {
                client.close(1000);
            });
            session.streamServer.close();
            session.streamServer.emit('close');
        };
        mockSessionRouter.post('/stop', async (req, res) => {
            await stopSession();
            res.json({ success: true });
        });
        // A pair of sockets, representing the 2-way connection between the session & WSs.
        // All websocket messages are written to wsSocket, and then read from sessionSocket
        // All session messages are written to sessionSocket, and then read from wsSocket and sent
        const { socket1: wsSocket, socket2: sessionSocket } = new DuplexPair();
        // This receives a lot of listeners! One channel per matcher, handler & completion checker,
        // and each adds listeners for data/error/finish/etc. That's OK, it's not generally a leak,
        // but maybe 100 would be a bit suspicious (unless you have 30+ active rules).
        sessionSocket.setMaxListeners(100);
        if (this.debug) {
            sessionSocket.on('data', (d) => {
                console.log('Streaming data from WS clients:', d.toString());
            });
            wsSocket.on('data', (d) => {
                console.log('Streaming data to WS clients:', d.toString());
            });
        }
        const streamServer = new Ws.Server({ noServer: true });
        streamServer.on('connection', (ws) => {
            let newClientStream = Ws.createWebSocketStream(ws, {});
            wsSocket.pipe(newClientStream).pipe(wsSocket, { end: false });
            const unpipe = () => {
                wsSocket.unpipe(newClientStream);
                newClientStream.unpipe(wsSocket);
            };
            newClientStream.on('error', unpipe);
            wsSocket.on('end', unpipe);
        });
        streamServer.on('close', () => {
            wsSocket.end();
            sessionSocket.end();
        });
        // Handle errors by logging & stopping this session
        const onStreamError = (e) => {
            if (!running)
                return; // We don't care about connection issues during shutdown
            console.error("Error in admin server stream, shutting down mock session");
            console.error(e);
            stopSession();
        };
        wsSocket.on('error', onStreamError);
        sessionSocket.on('error', onStreamError);
        const schema = (0, schema_1.makeExecutableSchema)({
            typeDefs: [
                AdminServer.baseSchema,
                ...Object.values(plugins).map(plugin => plugin.schema)
            ],
            resolvers: [
                this.buildBaseResolvers(sessionId),
                ...Object.values(plugins).map(plugin => plugin.buildResolvers(sessionSocket, this.ruleParams))
            ]
        });
        const subscriptionServer = subscriptions_transport_ws_1.SubscriptionServer.create({
            schema,
            execute: graphql_1.execute,
            subscribe: graphql_1.subscribe,
            keepAlive: this.webSocketKeepAlive
        }, {
            noServer: true
        });
        mockSessionRouter.use((0, express_1.createHandler)({
            schema,
            // Add console logging of all GQL errors:
            formatError: (error) => {
                console.error(error.stack);
                return error;
            }
        }));
        if (this.webSocketKeepAlive) {
            // If we have a keep-alive set, send the client a ping frame every Xms to
            // try and stop closes (especially by browsers) due to inactivity.
            const webSocketKeepAlive = setInterval(() => {
                [
                    ...streamServer.clients,
                    ...subscriptionServer.server.clients
                ].forEach((client) => {
                    if (client.readyState !== Ws.OPEN)
                        return;
                    client.ping();
                });
            }, this.webSocketKeepAlive);
            // We use the stream server's shutdown as an easy proxy event for full shutdown:
            streamServer.on('close', () => clearInterval(webSocketKeepAlive));
        }
        this.sessions[sessionId] = {
            sessionPlugins: plugins,
            router: mockSessionRouter,
            streamServer,
            subscriptionServer,
            stop: stopSession
        };
        this.eventEmitter.emit('mock-session-started', plugins, sessionId);
    }
    stop() {
        if (!this.server)
            return Promise.resolve();
        return Promise.all([
            this.server.destroy(),
        ].concat(Object.values(this.sessions).map((s) => s.stop()))).then(() => {
            this.server = null;
        });
    }
    buildBaseResolvers(sessionId) {
        return {
            Query: {
                ruleParameterKeys: () => this.ruleParameterKeys
            },
            Mutation: {
                reset: () => this.resetPluginsForSession(sessionId),
                enableDebug: () => this.enableDebugForSession(sessionId)
            },
            Raw: new graphql_1.GraphQLScalarType({
                name: 'Raw',
                description: 'A raw entity, serialized directly (must be JSON-compatible)',
                serialize: (value) => value,
                parseValue: (input) => input,
                parseLiteral: graphql_utils_1.parseAnyAst
            }),
            // Json exists just for API backward compatibility - all new data should be Raw.
            // Converting to JSON is pointless, since bodies all contain JSON anyway.
            Json: new graphql_1.GraphQLScalarType({
                name: 'Json',
                description: 'A JSON entity, serialized as a simple JSON string',
                serialize: (value) => JSON.stringify(value),
                parseValue: (input) => JSON.parse(input),
                parseLiteral: graphql_utils_1.parseAnyAst
            }),
            Void: new graphql_1.GraphQLScalarType({
                name: 'Void',
                description: 'Nothing at all',
                serialize: (value) => null,
                parseValue: (input) => null,
                parseLiteral: () => { throw new Error('Void literals are not supported'); }
            }),
            Buffer: new graphql_1.GraphQLScalarType({
                name: 'Buffer',
                description: 'A buffer',
                serialize: (value) => {
                    return value.toString('base64');
                },
                parseValue: (input) => {
                    return Buffer.from(input, 'base64');
                },
                parseLiteral: graphql_utils_1.parseAnyAst
            })
        };
    }
    ;
    resetPluginsForSession(sessionId) {
        return Promise.all(Object.values(this.sessions[sessionId].sessionPlugins).map(plugin => plugin.reset?.()));
    }
    enableDebugForSession(sessionId) {
        return Promise.all(Object.values(this.sessions[sessionId].sessionPlugins).map(plugin => plugin.enableDebug?.()));
    }
    get ruleParameterKeys() {
        return Object.keys(this.ruleParams);
    }
}
exports.AdminServer = AdminServer;
AdminServer.baseSchema = (0, graphql_tag_1.default) `
        type Mutation {
            reset: Void
            enableDebug: Void
        }

        type Query {
            ruleParameterKeys: [String!]!
        }

        type Subscription {
            _empty_placeholder_: Void # A placeholder so we can define an empty extendable type
        }

        scalar Void
        scalar Raw
        scalar Json
        scalar Buffer
    `;
//# sourceMappingURL=admin-server.js.map