// Copyright 2018-2022 the Deno authors. All rights reserved. MIT license.
// Copyright Joyent and Node contributors. All rights reserved. MIT license.
// deno-lint-ignore-file

import { addAbortSignalNoValidate } from "./add-abort-signal.js";
import { Buffer } from "buffer";
import { getDefaultHighWaterMark, getHighWaterMark } from "./state.js";
import { isUint8Array } from "../util/types.js";
import { Stream } from "./legacy.js";
import {
    ERR_INVALID_ARG_TYPE,
    ERR_METHOD_NOT_IMPLEMENTED,
    ERR_MULTIPLE_CALLBACK,
    ERR_STREAM_ALREADY_FINISHED,
    ERR_STREAM_CANNOT_PIPE,
    ERR_STREAM_DESTROYED,
    ERR_STREAM_NULL_VALUES,
    ERR_STREAM_WRITE_AFTER_END,
    ERR_UNKNOWN_ENCODING,
} from "../errors.js";
import destroyImpl from "./destroy.js";
import EE from "events";
import Readable from "./readable.js";

function _uint8ArrayToBuffer(chunk) {
    return Buffer.from(
        chunk.buffer,
        chunk.byteOffset,
        chunk.byteLength,
    );
}

const { errorOrDestroy } = destroyImpl;

// This function prevents a circular dependency with Duplex
// This checks if the passed stream is an instance of a Readable stream
// and one of its prototypes is named Duplex
function isDuplexStream(maybe_duplex) {
    const isReadable = Readable.prototype.isPrototypeOf(maybe_duplex);

    let prototype = maybe_duplex;
    let isDuplex = false;
    while (prototype?.constructor && prototype.constructor.name !== "Object") {
        if (prototype.constructor.name === "Duplex") {
            isDuplex = true;
            break;
        }
        prototype = Object.getPrototypeOf(prototype);
    }

    return isReadable && isDuplex;
}


function nop() { }

const kOnFinished = Symbol("kOnFinished");

function WritableState(options, stream, isDuplex) {
    // Duplex streams are both readable and writable, but share
    // the same options object.
    // However, some cases require setting options to different
    // values for the readable and the writable sides of the duplex stream,
    // e.g. options.readableObjectMode vs. options.writableObjectMode, etc.
    if (typeof isDuplex !== "boolean") {
        isDuplex = isDuplexStream(stream);
    }

    // Object stream flag to indicate whether or not this stream
    // contains buffers or objects.
    this.objectMode = !!(options && options.objectMode);

    if (isDuplex) {
        this.objectMode = this.objectMode ||
            !!(options && options.writableObjectMode);
    }

    // The point at which write() starts returning false
    // Note: 0 is a valid value, means that we always return false if
    // the entire buffer is not flushed immediately on write().
    this.highWaterMark = options
        ? getHighWaterMark(this, options, "writableHighWaterMark", isDuplex)
        : getDefaultHighWaterMark(false);

    // if _final has been called.
    this.finalCalled = false;

    // drain event flag.
    this.needDrain = false;
    // At the start of calling end()
    this.ending = false;
    // When end() has been called, and returned.
    this.ended = false;
    // When 'finish' is emitted.
    this.finished = false;

    // Has it been destroyed
    this.destroyed = false;

    // Should we decode strings into buffers before passing to _write?
    // this is here so that some node-core streams can optimize string
    // handling at a lower level.
    const noDecode = !!(options && options.decodeStrings === false);
    this.decodeStrings = !noDecode;

    // Crypto is kind of old and crusty.  Historically, its default string
    // encoding is 'binary' so we have to make this configurable.
    // Everything else in the universe uses 'utf8', though.
    this.defaultEncoding = (options && options.defaultEncoding) || "utf8";

    // Not an actual buffer we keep track of, but a measurement
    // of how much we're waiting to get pushed to some underlying
    // socket or file.
    this.length = 0;

    // A flag to see when we're in the middle of a write.
    this.writing = false;

    // When true all writes will be buffered until .uncork() call.
    this.corked = 0;

    // A flag to be able to tell if the onwrite cb is called immediately,
    // or on a later tick.  We set this to true at first, because any
    // actions that shouldn't happen until "later" should generally also
    // not happen before the first write call.
    this.sync = true;

    // A flag to know if we're processing previously buffered items, which
    // may call the _write() callback in the same tick, so that we don't
    // end up in an overlapped onwrite situation.
    this.bufferProcessing = false;

    // The callback that's passed to _write(chunk, cb).
    this.onwrite = onwrite.bind(undefined, stream);

    // The callback that the user supplies to write(chunk, encoding, cb).
    this.writecb = null;

    // The amount that is being written when _write is called.
    this.writelen = 0;

    // Storage for data passed to the afterWrite() callback in case of
    // synchronous _write() completion.
    this.afterWriteTickInfo = null;

    resetBuffer(this);

    // Number of pending user-supplied write callbacks
    // this must be 0 before 'finish' can be emitted.
    this.pendingcb = 0;

    // Stream is still being constructed and cannot be
    // destroyed until construction finished or failed.
    // Async construction is opt in, therefore we start as
    // constructed.
    this.constructed = true;

    // Emit prefinish if the only thing we're waiting for is _write cbs
    // This is relevant for synchronous Transform streams.
    this.prefinished = false;

    // True if the error was already emitted and should not be thrown again.
    this.errorEmitted = false;

    // Should close be emitted on destroy. Defaults to true.
    this.emitClose = !options || options.emitClose !== false;

    // Should .destroy() be called after 'finish' (and potentially 'end').
    this.autoDestroy = !options || options.autoDestroy !== false;

    // Indicates whether the stream has errored. When true all write() calls
    // should return false. This is needed since when autoDestroy
    // is disabled we need a way to tell whether the stream has failed.
    this.errored = null;

    // Indicates whether the stream has finished destroying.
    this.closed = false;

    // True if close has been emitted or would have been emitted
    // depending on emitClose.
    this.closeEmitted = false;

    this[kOnFinished] = [];
}

function resetBuffer(state) {
    state.buffered = [];
    state.bufferedIndex = 0;
    state.allBuffers = true;
    state.allNoop = true;
}

WritableState.prototype.getBuffer = function getBuffer() {
    return this.buffered.slice(this.bufferedIndex);
};

Object.defineProperty(WritableState.prototype, "bufferedRequestCount", {
    get() {
        return this.buffered.length - this.bufferedIndex;
    },
});

function Writable(options) {
    // Writable ctor is applied to Duplexes, too.
    // `realHasInstance` is necessary because using plain `instanceof`
    // would return false, as no `_writableState` property is attached.

    // Trying to use the custom `instanceof` for Writable here will also break the
    // Node.js LazyTransform implementation, which has a non-trivial getter for
    // `_writableState` that would lead to infinite recursion.

    // Checking for a Stream.Duplex instance is faster here instead of inside
    // the WritableState constructor, at least with V8 6.5.
    const isDuplex = isDuplexStream(this);

    if (
        !isDuplex && !Function.prototype[Symbol.hasInstance].call(Writable, this)
    ) {
        return new Writable(options);
    }

    this._writableState = new WritableState(options, this, isDuplex);

    if (options) {
        if (typeof options.write === "function") {
            this._write = options.write;
        }

        if (typeof options.writev === "function") {
            this._writev = options.writev;
        }

        if (typeof options.destroy === "function") {
            this._destroy = options.destroy;
        }

        if (typeof options.final === "function") {
            this._final = options.final;
        }

        if (typeof options.construct === "function") {
            this._construct = options.construct;
        }
        if (options.signal) {
            addAbortSignalNoValidate(options.signal, this);
        }
    }

    Stream.call(this, options);

    destroyImpl.construct(this, () => {
        const state = this._writableState;

        if (!state.writing) {
            clearBuffer(this, state);
        }

        finishMaybe(this, state);
    });
}

Object.setPrototypeOf(Writable.prototype, Stream.prototype);
Object.setPrototypeOf(Writable, Stream);

Object.defineProperty(Writable, Symbol.hasInstance, {
    value: function (object) {
        if (Function.prototype[Symbol.hasInstance].call(this, object)) return true;
        if (this !== Writable) return false;

        return object && object._writableState instanceof WritableState;
    },
});

// Otherwise people can pipe Writable streams, which is just wrong.
Writable.prototype.pipe = function () {
    errorOrDestroy(this, new ERR_STREAM_CANNOT_PIPE());
};

function _write(stream, chunk, encoding, cb) {
    const state = stream._writableState;

    if (typeof encoding === "function") {
        cb = encoding;
        encoding = state.defaultEncoding;
    } else {
        if (!encoding) {
            encoding = state.defaultEncoding;
        } else if (encoding !== "buffer" && !Buffer.isEncoding(encoding)) {
            throw new ERR_UNKNOWN_ENCODING(encoding);
        }
        if (typeof cb !== "function") {
            cb = nop;
        }
    }

    if (chunk === null) {
        throw new ERR_STREAM_NULL_VALUES();
    } else if (!state.objectMode) {
        if (typeof chunk === "string") {
            if (state.decodeStrings !== false) {
                chunk = Buffer.from(chunk, encoding);
                encoding = "buffer";
            }
        } else if (chunk instanceof Buffer) {
            encoding = "buffer";
        } else if (isUint8Array(chunk)) {
            chunk = _uint8ArrayToBuffer(chunk);
            encoding = "buffer";
        } else {
            throw new ERR_INVALID_ARG_TYPE(
                "chunk",
                ["string", "Buffer", "Uint8Array"],
                chunk,
            );
        }
    }

    let err;
    if (state.ending) {
        err = new ERR_STREAM_WRITE_AFTER_END();
    } else if (state.destroyed) {
        err = new ERR_STREAM_DESTROYED("write");
    }

    if (err) {
        nextTick(cb, err);
        errorOrDestroy(stream, err, true);
        return err;
    }
    state.pendingcb++;
    return writeOrBuffer(stream, state, chunk, encoding, cb);
}

Writable.prototype.write = function (chunk, encoding, cb) {
    return _write(this, chunk, encoding, cb) === true;
};

Writable.prototype.cork = function () {
    this._writableState.corked++;
};

Writable.prototype.uncork = function () {
    const state = this._writableState;

    if (state.corked) {
        state.corked--;

        if (!state.writing) {
            clearBuffer(this, state);
        }
    }
};

Writable.prototype.setDefaultEncoding = function setDefaultEncoding(encoding) {
    // node::ParseEncoding() requires lower case.
    if (typeof encoding === "string") {
        encoding = encoding.toLowerCase();
    }
    if (!Buffer.isEncoding(encoding)) {
        throw new ERR_UNKNOWN_ENCODING(encoding);
    }
    this._writableState.defaultEncoding = encoding;
    return this;
};

// If we're already writing something, then just put this
// in the queue, and wait our turn.  Otherwise, call _write
// If we return false, then we need a drain event, so set that flag.
function writeOrBuffer(stream, state, chunk, encoding, callback) {
    const len = state.objectMode ? 1 : chunk.length;

    state.length += len;

    // stream._write resets state.length
    const ret = state.length < state.highWaterMark;
    // We must ensure that previous needDrain will not be reset to false.
    if (!ret) {
        state.needDrain = true;
    }

    if (state.writing || state.corked || state.errored || !state.constructed) {
        state.buffered.push({ chunk, encoding, callback });
        if (state.allBuffers && encoding !== "buffer") {
            state.allBuffers = false;
        }
        if (state.allNoop && callback !== nop) {
            state.allNoop = false;
        }
    } else {
        state.writelen = len;
        state.writecb = callback;
        state.writing = true;
        state.sync = true;
        stream._write(chunk, encoding, state.onwrite);
        state.sync = false;
    }

    // Return false if errored or destroyed in order to break
    // any synchronous while(stream.write(data)) loops.
    return ret && !state.errored && !state.destroyed;
}

function doWrite(stream, state, writev, len, chunk, encoding, cb) {
    state.writelen = len;
    state.writecb = cb;
    state.writing = true;
    state.sync = true;
    if (state.destroyed) {
        state.onwrite(new ERR_STREAM_DESTROYED("write"));
    } else if (writev) {
        stream._writev(chunk, state.onwrite);
    } else {
        stream._write(chunk, encoding, state.onwrite);
    }
    state.sync = false;
}

function onwriteError(stream, state, er, cb) {
    --state.pendingcb;

    cb(er);
    // Ensure callbacks are invoked even when autoDestroy is
    // not enabled. Passing `er` here doesn't make sense since
    // it's related to one specific write, not to the buffered
    // writes.
    errorBuffer(state);
    // This can emit error, but error must always follow cb.
    errorOrDestroy(stream, er);
}

function onwrite(stream, er) {
    const state = stream._writableState;
    const sync = state.sync;
    const cb = state.writecb;

    if (typeof cb !== "function") {
        errorOrDestroy(stream, new ERR_MULTIPLE_CALLBACK());
        return;
    }

    state.writing = false;
    state.writecb = null;
    state.length -= state.writelen;
    state.writelen = 0;

    if (er) {
        // Avoid V8 leak, https://github.com/nodejs/node/pull/34103#issuecomment-652002364
        er.stack; // eslint-disable-line no-unused-expressions

        if (!state.errored) {
            state.errored = er;
        }

        // In case of duplex streams we need to notify the readable side of the
        // error.
        if (stream._readableState && !stream._readableState.errored) {
            stream._readableState.errored = er;
        }

        if (sync) {
            nextTick(onwriteError, stream, state, er, cb);
        } else {
            onwriteError(stream, state, er, cb);
        }
    } else {
        if (state.buffered.length > state.bufferedIndex) {
            clearBuffer(stream, state);
        }

        if (sync) {
            // It is a common case that the callback passed to .write() is always
            // the same. In that case, we do not schedule a new nextTick(), but
            // rather just increase a counter, to improve performance and avoid
            // memory allocations.
            if (
                state.afterWriteTickInfo !== null &&
                state.afterWriteTickInfo.cb === cb
            ) {
                state.afterWriteTickInfo.count++;
            } else {
                state.afterWriteTickInfo = { count: 1, cb, stream, state };
                nextTick(afterWriteTick, state.afterWriteTickInfo);
            }
        } else {
            afterWrite(stream, state, 1, cb);
        }
    }
}

function afterWriteTick({ stream, state, count, cb }) {
    state.afterWriteTickInfo = null;
    return afterWrite(stream, state, count, cb);
}

function afterWrite(stream, state, count, cb) {
    const needDrain = !state.ending && !stream.destroyed && state.length === 0 &&
        state.needDrain;
    if (needDrain) {
        state.needDrain = false;
        stream.emit("drain");
    }

    while (count-- > 0) {
        state.pendingcb--;
        cb();
    }

    if (state.destroyed) {
        errorBuffer(state);
    }

    finishMaybe(stream, state);
}

// If there's something in the buffer waiting, then invoke callbacks.
function errorBuffer(state) {
    if (state.writing) {
        return;
    }

    for (let n = state.bufferedIndex; n < state.buffered.length; ++n) {
        const { chunk, callback } = state.buffered[n];
        const len = state.objectMode ? 1 : chunk.length;
        state.length -= len;
        callback(new ERR_STREAM_DESTROYED("write"));
    }

    const onfinishCallbacks = state[kOnFinished].splice(0);
    for (let i = 0; i < onfinishCallbacks.length; i++) {
        onfinishCallbacks[i](new ERR_STREAM_DESTROYED("end"));
    }

    resetBuffer(state);
}

// If there's something in the buffer waiting, then process it.
function clearBuffer(stream, state) {
    if (
        state.corked ||
        state.bufferProcessing ||
        state.destroyed ||
        !state.constructed
    ) {
        return;
    }

    const { buffered, bufferedIndex, objectMode } = state;
    const bufferedLength = buffered.length - bufferedIndex;

    if (!bufferedLength) {
        return;
    }

    let i = bufferedIndex;

    state.bufferProcessing = true;
    if (bufferedLength > 1 && stream._writev) {
        state.pendingcb -= bufferedLength - 1;

        const callback = state.allNoop ? nop : (err) => {
            for (let n = i; n < buffered.length; ++n) {
                buffered[n].callback(err);
            }
        };
        // Make a copy of `buffered` if it's going to be used by `callback` above,
        // since `doWrite` will mutate the array.
        const chunks = state.allNoop && i === 0 ? buffered : buffered.slice(i);
        chunks.allBuffers = state.allBuffers;

        doWrite(stream, state, true, state.length, chunks, "", callback);

        resetBuffer(state);
    } else {
        do {
            const { chunk, encoding, callback } = buffered[i];
            buffered[i++] = null;
            const len = objectMode ? 1 : chunk.length;
            doWrite(stream, state, false, len, chunk, encoding, callback);
        } while (i < buffered.length && !state.writing);

        if (i === buffered.length) {
            resetBuffer(state);
        } else if (i > 256) {
            buffered.splice(0, i);
            state.bufferedIndex = 0;
        } else {
            state.bufferedIndex = i;
        }
    }
    state.bufferProcessing = false;
}

Writable.prototype._write = function (chunk, encoding, cb) {
    if (this._writev) {
        this._writev([{ chunk, encoding }], cb);
    } else {
        throw new ERR_METHOD_NOT_IMPLEMENTED("_write()");
    }
};

Writable.prototype._writev = null;

Writable.prototype.end = function (chunk, encoding, cb) {
    const state = this._writableState;

    if (typeof chunk === "function") {
        cb = chunk;
        chunk = null;
        encoding = null;
    } else if (typeof encoding === "function") {
        cb = encoding;
        encoding = null;
    }

    let err;

    if (chunk !== null && chunk !== undefined) {
        const ret = _write(this, chunk, encoding);
        if (ret instanceof Error) {
            err = ret;
        }
    }

    // .end() fully uncorks.
    if (state.corked) {
        state.corked = 1;
        this.uncork();
    }

    if (err) {
        // Do nothing...
    } else if (!state.errored && !state.ending) {
        // This is forgiving in terms of unnecessary calls to end() and can hide
        // logic errors. However, usually such errors are harmless and causing a
        // hard error can be disproportionately destructive. It is not always
        // trivial for the user to determine whether end() needs to be called
        // or not.

        state.ending = true;
        finishMaybe(this, state, true);
        state.ended = true;
    } else if (state.finished) {
        err = new ERR_STREAM_ALREADY_FINISHED("end");
    } else if (state.destroyed) {
        err = new ERR_STREAM_DESTROYED("end");
    }

    if (typeof cb === "function") {
        if (err || state.finished) {
            nextTick(cb, err);
        } else {
            state[kOnFinished].push(cb);
        }
    }

    return this;
};

function needFinish(state) {
    return (state.ending &&
        state.constructed &&
        state.length === 0 &&
        !state.errored &&
        state.buffered.length === 0 &&
        !state.finished &&
        !state.writing &&
        !state.errorEmitted &&
        !state.closeEmitted);
}

function callFinal(stream, state) {
    let called = false;

    function onFinish(err) {
        if (called) {
            errorOrDestroy(stream, err ?? ERR_MULTIPLE_CALLBACK());
            return;
        }
        called = true;

        state.pendingcb--;
        if (err) {
            const onfinishCallbacks = state[kOnFinished].splice(0);
            for (let i = 0; i < onfinishCallbacks.length; i++) {
                onfinishCallbacks[i](err);
            }
            errorOrDestroy(stream, err, state.sync);
        } else if (needFinish(state)) {
            state.prefinished = true;
            stream.emit("prefinish");
            // Backwards compat. Don't check state.sync here.
            // Some streams assume 'finish' will be emitted
            // asynchronously relative to _final callback.
            state.pendingcb++;
            nextTick(finish, stream, state);
        }
    }

    state.sync = true;
    state.pendingcb++;

    try {
        const result = stream._final(onFinish);
        if (result != null) {
            const then = result.then;
            if (typeof then === "function") {
                then.call(
                    result,
                    function () {
                        nextTick(onFinish, null);
                    },
                    function (err) {
                        nextTick(onFinish, err);
                    },
                );
            }
        }
    } catch (err) {
        onFinish(stream, state, err);
    }

    state.sync = false;
}

function prefinish(stream, state) {
    if (!state.prefinished && !state.finalCalled) {
        if (typeof stream._final === "function" && !state.destroyed) {
            state.finalCalled = true;
            callFinal(stream, state);
        } else {
            state.prefinished = true;
            stream.emit("prefinish");
        }
    }
}

function finishMaybe(stream, state, sync) {
    if (needFinish(state)) {
        prefinish(stream, state);
        if (state.pendingcb === 0 && needFinish(state)) {
            state.pendingcb++;
            if (sync) {
                nextTick(finish, stream, state);
            } else {
                finish(stream, state);
            }
        }
    }
}

function finish(stream, state) {
    state.pendingcb--;
    state.finished = true;

    const onfinishCallbacks = state[kOnFinished].splice(0);
    for (let i = 0; i < onfinishCallbacks.length; i++) {
        onfinishCallbacks[i]();
    }

    stream.emit("finish");

    if (state.autoDestroy) {
        // In case of duplex streams we need a way to detect
        // if the readable side is ready for autoDestroy as well.
        const rState = stream._readableState;
        const autoDestroy = !rState || (
            rState.autoDestroy &&
            // We don't expect the readable to ever 'end'
            // if readable is explicitly set to false.
            (rState.endEmitted || rState.readable === false)
        );
        if (autoDestroy) {
            stream.destroy();
        }
    }
}

Object.defineProperties(Writable.prototype, {
    destroyed: {
        get() {
            return this._writableState ? this._writableState.destroyed : false;
        },
        set(value) {
            // Backward compatibility, the user is explicitly managing destroyed.
            if (this._writableState) {
                this._writableState.destroyed = value;
            }
        },
    },

    writable: {
        get() {
            const w = this._writableState;
            // w.writable === false means that this is part of a Duplex stream
            // where the writable side was disabled upon construction.
            // Compat. The user might manually disable writable side through
            // deprecated setter.
            return !!w && w.writable !== false && !w.destroyed && !w.errored &&
                !w.ending && !w.ended;
        },
        set(val) {
            // Backwards compatible.
            if (this._writableState) {
                this._writableState.writable = !!val;
            }
        },
    },

    writableFinished: {
        get() {
            return this._writableState ? this._writableState.finished : false;
        },
    },

    writableObjectMode: {
        get() {
            return this._writableState ? this._writableState.objectMode : false;
        },
    },

    writableBuffer: {
        get() {
            return this._writableState && this._writableState.getBuffer();
        },
    },

    writableEnded: {
        get() {
            return this._writableState ? this._writableState.ending : false;
        },
    },

    writableNeedDrain: {
        get() {
            const wState = this._writableState;
            if (!wState) return false;
            return !wState.destroyed && !wState.ending && wState.needDrain;
        },
    },

    writableHighWaterMark: {
        get() {
            return this._writableState && this._writableState.highWaterMark;
        },
    },

    writableCorked: {
        get() {
            return this._writableState ? this._writableState.corked : 0;
        },
    },

    writableLength: {
        get() {
            return this._writableState && this._writableState.length;
        },
    },
});

const destroy = destroyImpl.destroy;
Writable.prototype.destroy = function (err, cb) {
    const state = this._writableState;

    // Invoke pending callbacks.
    if (
        !state.destroyed &&
        (state.bufferedIndex < state.buffered.length ||
            state[kOnFinished].length)
    ) {
        nextTick(errorBuffer, state);
    }

    destroy.call(this, err, cb);
    return this;
};

Writable.prototype._undestroy = destroyImpl.undestroy;
Writable.prototype._destroy = function (err, cb) {
    cb(err);
};

Writable.prototype[EE.captureRejectionSymbol] = function (err) {
    this.destroy(err);
};

Writable.WritableState = WritableState;

export default Writable;
export { Writable, WritableState };