"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.ImapStream = void 0;
const node_stream_1 = require("node:stream");
const logger_js_1 = require("../logger.js");
const limits_js_1 = require("./limits.js");
const errors_js_1 = require("../errors.js");
const LINE = 0x01;
const LITERAL = 0x02;
const LF = 0x0a;
const CR = 0x0d;
const NUM_0 = 0x30;
const NUM_9 = 0x39;
const CURLY_OPEN = 0x7b;
const CURLY_CLOSE = 0x7d;
/**
 * A Transform stream that parses raw IMAP protocol data from a socket into structured
 * command/response objects. Reads binary input, splits it into lines delimited by LF,
 * extracts literal data blocks based on IMAP literal size markers (e.g., "{123}\r\n"),
 * and emits each complete command as a readable object containing the payload Buffer
 * and any associated literal Buffers. Enforces a maximum literal size of 1GB.
 */
class ImapStream extends node_stream_1.Transform {
    /**
     * Creates a new ImapStream instance.
     *
     * @param options - Stream options, see ImapStreamOptions.
     */
    constructor(options) {
        super({
            //writableHighWaterMark: 3,
            readableObjectMode: true,
            writableObjectMode: false
        });
        this.options = options || {};
        this.cid = this.options.cid;
        if (this.options.logger && typeof this.options.logger === 'object') {
            this.log = this.options.logger;
        }
        else {
            this.log = (0, logger_js_1.createConnectionLogger)({ cid: this.cid, logRaw: this.options.logRaw });
        }
        this.readBytesCounter = 0;
        // Maximum length of a single line (response without a literal). Bounds the line buffer
        // so a server that never sends a line terminator cannot exhaust memory.
        this.maxLineLength = (0, limits_js_1.normalizeLimit)(this.options.maxLineLength, limits_js_1.MAX_LINE_SIZE);
        // Maximum size of a single literal block. Bounds peak memory allocation so a server
        // announcing an oversized literal cannot exhaust memory.
        this.maxLiteralSize = (0, limits_js_1.normalizeLimit)(this.options.maxLiteralSize, limits_js_1.MAX_LITERAL_SIZE);
        this.maxResponseSize = (0, limits_js_1.normalizeLimit)(this.options.maxResponseSize, limits_js_1.MAX_RESPONSE_SIZE);
        this.state = LINE;
        this.literalWaiting = 0;
        this.inputBuffer = []; // lines
        this.lineBuffer = []; // current line
        this.lineBytes = 0; // bytes currently buffered for the in-progress line
        this.literalBuffer = [];
        this.literals = [];
        this.responseBytes = 0; // bytes accumulated for the in-progress response (lines + declared literals)
        this.compress = false;
        this.secureConnection = this.options.secureConnection;
        this.processingInput = false;
        this.inputQueue = []; // unprocessed input chunks
        this.activeInput = null; // chunk currently being processed (already shifted off inputQueue)
        // Resolver of the in-flight push() backpressure promise, so destruction can settle it
        // instead of leaving processInput() awaiting a consumer that will never read again.
        this.pendingPush = null;
    }
    /**
     * Terminally fails the stream. Used for response limit violations and for any other
     * error raised while parsing.
     *
     * The stream is destroyed instead of only emitting `error`: emitting on a Transform leaves
     * it running, so the caller would keep scanning the rejected payload and could emit it as
     * protocol (an oversized literal body contains attacker-chosen CRLF delimited lines).
     * Destroying stops all parsing, drops the offending line, and releases every queued
     * transform callback exactly once (see `_destroy()`).
     *
     * `destroyed` (set synchronously by destroy()) is the single liveness flag every other path
     * checks, so a second failure attempt is a no-op and nothing is parsed after the first.
     *
     * @param err - The error to destroy the stream with.
     * @returns Always false, so callers can `return this.failStream(err)`.
     */
    failStream(err) {
        if (this.destroyed) {
            return false;
        }
        this.destroy(err);
        return false;
    }
    /**
     * Releases a queued input chunk's transform callback exactly once, signalling the writable
     * side that the chunk was consumed. The mirror image of ImapFlow's releaseStreamData(), which
     * releases the readable items this stream pushes downstream.
     *
     * @param item - Queue entry holding the chunk and its transform callback.
     */
    releaseInput(item) {
        if (!item || item.released) {
            return;
        }
        item.released = true;
        if (typeof item.next === 'function') {
            item.next();
        }
    }
    /**
     * Checks whether the given line buffer ends with an IMAP literal size marker
     * (e.g., "{123}\r\n"). If a valid marker is found and the literal size is within
     * the allowed maximum, switches the stream state to LITERAL mode and records
     * the expected number of literal bytes.
     *
     * @param line - The line buffer to check for a trailing literal marker.
     * @returns True if a valid literal marker was found and literal state was activated, false otherwise.
     */
    checkLiteralMarker(line) {
        if (!line || !line.length) {
            return false;
        }
        let pos = line.length - 1;
        if (line[pos] !== LF) {
            return false;
        }
        pos--;
        if (pos >= 0 && line[pos] === CR) {
            pos--;
        }
        if (pos < 0 || !pos || line[pos] !== CURLY_CLOSE) {
            return false;
        }
        pos--;
        // Scan backwards through the line to find an IMAP literal marker: {size}\r\n
        // The format is: '{' followed by one or more ASCII digits followed by '}'.
        // Only the digit run's bounds are tracked - a single linear pass, unlike
        // collecting digits into a growing array, which would make a line of n digits
        // cost O(n^2). The run length is deliberately not capped: the RFC "number"
        // production permits leading zeros, so a long digit run can still denote a
        // small, valid size, and treating the marker as an ordinary line instead
        // would feed the announced literal body to the line parser and desynchronize
        // the session.
        let digitsEnd = pos;
        for (; pos >= 0; pos--) {
            let c = line[pos];
            if (c >= NUM_0 && c <= NUM_9) {
                continue;
            }
            if (c === CURLY_OPEN && pos < digitsEnd) {
                // Skip leading zeros so only the significant digits are converted: a
                // marker padded with megabytes of zeros must not cost a string
                // allocation and Number() parse of the whole run.
                let digitsStart = pos + 1;
                while (digitsStart < digitsEnd && line[digitsStart] === NUM_0) {
                    digitsStart++;
                }
                // More significant digits than any number64 has cannot fit any
                // permissible maxLiteralSize; fail closed without materializing them
                if (digitsEnd + 1 - digitsStart > 19) {
                    return this.failStream((0, limits_js_1.createLiteralTooLargeError)(Infinity, this.maxLiteralSize, 'the widest permissible literal size (19 digits)'));
                }
                const literalSize = Number(line.toString('latin1', digitsStart, digitsEnd + 1));
                if (literalSize > this.maxLiteralSize) {
                    return this.failStream((0, limits_js_1.createLiteralTooLargeError)(literalSize, this.maxLiteralSize));
                }
                this.state = LITERAL;
                this.literalWaiting = literalSize;
                return true;
            }
            return false;
        }
        return false;
    }
    /**
     * Enforces the configured line-length cap for a projected line length. The projected length
     * covers every byte of the line, the line terminator included, whether or not the line was
     * split across input chunks. A line exactly at the limit is accepted.
     *
     * @param lineLength - Total length the current line would reach.
     * @returns True if the line is within the limit, false if the stream was failed.
     */
    checkLineLength(lineLength) {
        if (lineLength <= this.maxLineLength) {
            return true;
        }
        const err = (0, errors_js_1.createImapError)(`Line length ${lineLength} exceeds maximum allowed size of ${this.maxLineLength} bytes`, 'LineTooLarge', {
            lineLength,
            maxSize: this.maxLineLength
        });
        return this.failStream(err);
    }
    /**
     * Enforces the configured per-response size cap: the cumulative bytes of every line
     * segment and declared literal of the response currently being assembled. Counting
     * declared literal sizes at marker time means an oversized total is rejected before
     * the literal bytes even arrive. The counter is reset when a response is emitted.
     *
     * @param additionalBytes - Bytes the next token would add to the response.
     * @param peek - Measure only, without committing the bytes to the counter.
     *   Used for a line that is still being assembled: its bytes are committed once, when the
     *   line completes.
     * @returns True if within the limit, false if the stream was failed.
     */
    checkResponseSize(additionalBytes, peek) {
        let total = this.responseBytes + additionalBytes;
        if (total <= this.maxResponseSize) {
            if (!peek) {
                this.responseBytes = total;
            }
            return true;
        }
        const err = (0, errors_js_1.createImapError)(`Response size ${total} exceeds maximum allowed size of ${this.maxResponseSize} bytes`, 'ResponseTooLarge', {
            responseSize: total,
            maxSize: this.maxResponseSize
        });
        return this.failStream(err);
    }
    /**
     * Processes a single input chunk of raw data. In LINE state, scans for LF-terminated
     * lines and checks for literal markers. In LITERAL state, collects the expected number
     * of literal bytes. When a complete command (with all its literals) is assembled, it is
     * pushed downstream as a readable object.
     *
     * @param chunk - The raw data chunk to process.
     */
    async processInputChunk(chunk) {
        // Every state switch hands back the offset to resume from instead of recursing, so a
        // chunk packed with thousands of small literals can not exhaust the call stack
        let nextPos = 0;
        while (nextPos !== null) {
            nextPos = await this.processChunkSegment(chunk, nextPos);
        }
    }
    /**
     * Processes the chunk from `startPos` until the parser state changes or the chunk ends.
     *
     * @returns The offset to continue from after a state switch, or `null` when done with the chunk.
     */
    async processChunkSegment(chunk, startPos) {
        if (this.destroyed || startPos >= chunk.length) {
            return null;
        }
        switch (this.state) {
            case LINE: {
                let lineStart = startPos;
                for (let i = startPos, len = chunk.length; i < len; i++) {
                    if (chunk[i] === LF) {
                        // line end found. Measure the completed line (terminator included) before
                        // concatenating or emitting anything, so the cap does not depend on where
                        // TCP chunk boundaries happen to fall.
                        let segment = chunk.subarray(lineStart, i + 1);
                        if (!this.checkLineLength(this.lineBytes + segment.length)) {
                            return null;
                        }
                        this.lineBuffer.push(segment);
                        lineStart = i + 1;
                        let line = this.lineBuffer.length === 1 ? this.lineBuffer[0] : Buffer.concat(this.lineBuffer);
                        this.lineBuffer = [];
                        this.lineBytes = 0;
                        // try to detect if this is a literal start. An oversized literal fails the
                        // stream, so the marker line must not be buffered before the check - it
                        // would otherwise be emitted as part of the rejected command.
                        let isLiteralMarker = this.checkLiteralMarker(line);
                        if (this.destroyed) {
                            return null;
                        }
                        // Count the line itself and, for a literal marker, the declared
                        // literal bytes against the cumulative per-response budget, so a
                        // response assembled from many tokens stays bounded as a whole
                        if (!this.checkResponseSize(line.length + (isLiteralMarker ? this.literalWaiting : 0))) {
                            return null;
                        }
                        this.inputBuffer.push(line);
                        if (isLiteralMarker) {
                            // switch into literal mode and start over
                            return lineStart;
                        }
                        // reached end of command input, emit it
                        let payload = this.inputBuffer.length === 1 ? this.inputBuffer[0] : Buffer.concat(this.inputBuffer);
                        let literals = this.literals;
                        this.inputBuffer = [];
                        this.literals = [];
                        this.responseBytes = 0;
                        if (payload.length) {
                            // remove final line terminator (\n or \r\n)
                            if (payload[payload.length - 1] === LF) {
                                let end = payload.length - 1;
                                if (end > 0 && payload[end - 1] === CR) {
                                    end--;
                                }
                                payload = payload.subarray(0, end);
                            }
                            if (payload.length) {
                                // Whether more buffered input already followed this command on the
                                // wire - more bytes in this chunk or another queued chunk. Captured
                                // per emitted command (immutable on the pushed object) so a later
                                // command cannot overwrite it; consumers that care about pipelining
                                // boundaries can read it from the pushed object.
                                let trailingAfterLine = lineStart < chunk.length || this.inputQueue.length > 0;
                                await new Promise(resolve => {
                                    // Tracked so destruction can settle the wait instead of leaving
                                    // this loop (and the chunk's transform callback) pending forever
                                    // when the consumer stops reading.
                                    this.pendingPush = resolve;
                                    const item = { payload, literals, next: resolve, trailingAfterLine };
                                    this.push(item);
                                });
                                this.pendingPush = null;
                                if (this.destroyed) {
                                    return null;
                                }
                            }
                        }
                    }
                }
                if (lineStart < chunk.length) {
                    // No line terminator was found in the remaining bytes; carry the tail over to
                    // the next chunk after measuring the line it belongs to.
                    let tail = chunk.subarray(lineStart);
                    // The response counter is only committed when a line completes, so an
                    // in-progress line is measured against the remaining budget separately.
                    // Without this a response cap lowered to bound parser memory buys nothing
                    // while a server streams a line that never terminates - only the much
                    // larger line cap would hold it back.
                    if (!this.checkLineLength(this.lineBytes + tail.length) || !this.checkResponseSize(this.lineBytes + tail.length, true)) {
                        return null;
                    }
                    this.lineBytes += tail.length;
                    this.lineBuffer.push(tail);
                }
                break;
            }
            case LITERAL: {
                const remainingInChunk = chunk.length - startPos;
                const bytesToRead = Math.min(remainingInChunk, this.literalWaiting);
                const partial = startPos === 0 && bytesToRead === chunk.length ? chunk : chunk.subarray(startPos, startPos + bytesToRead);
                this.literalBuffer.push(partial);
                this.literalWaiting -= bytesToRead;
                if (this.literalWaiting === 0) {
                    this.literals.push(Buffer.concat(this.literalBuffer));
                    this.literalBuffer = [];
                    this.state = LINE;
                    if (remainingInChunk > bytesToRead) {
                        return startPos + bytesToRead;
                    }
                }
                break;
            }
        }
        return null;
    }
    /**
     * Drains the input queue by processing each queued chunk sequentially.
     * Yields to the event loop every 10 chunks to prevent CPU blocking on
     * large bursts of incoming data.
     *
     * The `processingInput` guard is cleared in the same synchronous step that finds the queue
     * empty. Clearing it later (in a promise handler) leaves a gap of a few microtasks where a
     * chunk delivered by the writable side is queued but no loop is started for it, so its
     * transform callback is never called and the socket is never read again. Workers deliver
     * the next chunk inside that gap.
     */
    async processInput() {
        try {
            let data;
            let processedCount = 0;
            while (!this.destroyed && (data = this.inputQueue.shift())) {
                this.activeInput = data;
                await this.processInputChunk(data.chunk);
                this.activeInput = null;
                // mark chunk as processed
                this.releaseInput(data);
                // Yield to event loop every 10 chunks to prevent CPU blocking
                processedCount++;
                if (processedCount % 10 === 0) {
                    await new Promise(resolve => setImmediate(resolve));
                }
            }
        }
        finally {
            this.processingInput = false;
        }
    }
    /**
     * Transform stream implementation. Receives raw data chunks from the writable side,
     * converts strings to Buffers, tracks total bytes read, optionally logs raw data,
     * and queues the chunk for asynchronous processing.
     *
     * @param chunk - The incoming data chunk.
     * @param encoding - The encoding if chunk is a string.
     * @param next - Callback to signal that this chunk has been consumed.
     */
    _transform(chunk, encoding, next) {
        if (typeof chunk === 'string') {
            chunk = Buffer.from(chunk, encoding);
        }
        if (!chunk || !chunk.length) {
            return next();
        }
        this.readBytesCounter += chunk.length;
        if (this.options.logRaw) {
            this.log.trace({
                src: 's',
                msg: 'read from socket',
                data: chunk.toString('base64'),
                compress: !!this.compress,
                secure: !!this.secureConnection,
                cid: this.cid
            });
        }
        // A terminal parser failure must not accept any more protocol input, even if the
        // transport delivers a chunk that was already in flight.
        if (this.destroyed) {
            return next();
        }
        // Queue the chunk for async processing. The 'next' callback serves as
        // backpressure: it is called only after this chunk is fully processed,
        // which signals the writable side that more data can be accepted.
        this.inputQueue.push({ chunk, next });
        if (!this.processingInput) {
            this.processingInput = true;
            this.processInput().catch(err => this.failStream(err));
        }
    }
    /**
     * Flush implementation called when the writable side ends. Signals completion immediately.
     *
     * @param next - Callback to signal flush completion.
     */
    _flush(next) {
        next();
    }
    /**
     * Destroy implementation for cleanup. Clears all internal buffers, drains the input queue
     * by invoking pending callbacks, and forwards the error (if any) to the callback.
     *
     * @param err - The error that caused destruction, or null.
     * @param callback - Callback to signal destruction completion.
     */
    _destroy(err, callback) {
        // Destruction is the single release point for parser-owned callbacks, so a terminal
        // failure can never leave the writable side or the processing loop waiting.
        this.inputBuffer = [];
        this.lineBuffer = [];
        this.lineBytes = 0;
        this.literalBuffer = [];
        this.literals = [];
        this.responseBytes = 0;
        // Settle an in-flight push() wait so processInput() can unwind
        if (typeof this.pendingPush === 'function') {
            const resolve = this.pendingPush;
            this.pendingPush = null;
            resolve();
        }
        // Release the chunk currently being processed, then everything still queued.
        // releaseInput() is idempotent, so the processing loop releasing the same chunk
        // afterwards is a no-op.
        this.releaseInput(this.activeInput);
        this.activeInput = null;
        while (this.inputQueue.length) {
            this.releaseInput(this.inputQueue.shift());
        }
        callback(err);
    }
}
exports.ImapStream = ImapStream;