/* -*- Mode: C++; tab-width: 8; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
/* vim: set ts=8 sts=2 et sw=2 tw=80: */
/* This Source Code Form is subject to the terms of the Mozilla Public
 * License, v. 2.0. If a copy of the MPL was not distributed with this
 * file, You can obtain one at http://mozilla.org/MPL/2.0/. */

#include "mozilla/dom/cache/ReadStream.h"

#include "mozilla/Unused.h"
#include "mozilla/dom/cache/CacheStreamControlChild.h"
#include "mozilla/dom/cache/CacheStreamControlParent.h"
#include "mozilla/dom/cache/CacheTypes.h"
#include "mozilla/ipc/IPCStreamUtils.h"
#include "mozilla/SnappyUncompressInputStream.h"
#include "nsIAsyncInputStream.h"
#include "nsIThread.h"
#include "nsStringStream.h"
#include "nsTArray.h"

namespace mozilla::dom::cache {

using mozilla::Unused;

// ----------------------------------------------------------------------------

// The inner stream class.  This is where all of the real work is done.  As
// an invariant Inner::Close() must be called before ~Inner().  This is
// guaranteed by our outer ReadStream class.
class ReadStream::Inner final : public ReadStream::Controllable {
 public:
  Inner(StreamControl* aControl, const nsID& aId, nsIInputStream* aStream);

  void Serialize(Maybe<CacheReadStream>* aReadStreamOut, ErrorResult& aRv);

  void Serialize(CacheReadStream* aReadStreamOut, ErrorResult& aRv);

  // ReadStream::Controllable methods
  virtual void CloseStream() override;

  virtual void CloseStreamWithoutReporting() override;

  virtual bool HasEverBeenRead() const override;

  // Simulate nsIInputStream methods, but we don't actually inherit from it
  nsresult Close();

  nsresult Available(uint64_t* aNumAvailableOut);

  nsresult StreamStatus();

  nsresult Read(char* aBuf, uint32_t aCount, uint32_t* aNumReadOut);

  nsresult ReadSegments(nsWriteSegmentFun aWriter, void* aClosure,
                        uint32_t aCount, uint32_t* aNumReadOut);

  nsresult IsNonBlocking(bool* aNonBlockingOut);

  NS_DECL_OWNINGTHREAD;

  ~Inner();

 private:
  class NoteClosedRunnable;
  class ForgetRunnable;

  void NoteClosed();

  void Forget();

  void NoteClosedOnOwningThread();

  void ForgetOnOwningThread();

  nsIInputStream* EnsureStream();

  void AsyncOpenStreamOnOwningThread();

  void MaybeAbortAsyncOpenStream();

  void OpenStreamFailed();

  inline SafeRefPtr<Inner> SafeRefPtrFromThis() {
    return Controllable::SafeRefPtrFromThis().downcast<Inner>();
  }

  // Weak ref to the stream control actor.  The actor will always call either
  // CloseStream() or CloseStreamWithoutReporting() before it's destroyed.  The
  // weak ref is cleared in the resulting NoteClosedOnOwningThread() or
  // ForgetOnOwningThread() method call.
  StreamControl* mControl;

  const nsID mId;
  nsCOMPtr<nsISerialEventTarget> mOwningEventTarget;

  enum State { Open, Closed, NumStates };
  Atomic<State> mState;
  Atomic<bool> mHasEverBeenRead;
  bool mAsyncOpenStarted;

  // The wrapped stream objects may not be threadsafe.  We need to be able
  // to close a stream on our owning thread while an IO thread is simultaneously
  // reading the same stream.  Therefore, protect all access to these stream
  // objects with a mutex.
  Mutex mMutex MOZ_UNANNOTATED;
  CondVar mCondVar;
  nsCOMPtr<nsIInputStream> mStream;
  nsCOMPtr<nsIInputStream> mSnappyStream;
};

// ----------------------------------------------------------------------------

// Runnable to notify actors that the ReadStream has closed.  This must
// be done on the thread associated with the PBackground actor.  Must be
// cancelable to execute on Worker threads (which can occur when the
// ReadStream is constructed on a child process Worker thread).
class ReadStream::Inner::NoteClosedRunnable final : public CancelableRunnable {
 public:
  explicit NoteClosedRunnable(SafeRefPtr<ReadStream::Inner> aStream)
      : CancelableRunnable("dom::cache::ReadStream::Inner::NoteClosedRunnable"),
        mStream(std::move(aStream)) {}

  NS_IMETHOD Run() override {
    mStream->NoteClosedOnOwningThread();
    return NS_OK;
  }

  // Note, we must proceed with the Run() method since our actor will not
  // clean itself up until we note that the stream is closed.
  nsresult Cancel() override {
    Run();
    return NS_OK;
  }

 private:
  ~NoteClosedRunnable() = default;

  const SafeRefPtr<ReadStream::Inner> mStream;
};

// ----------------------------------------------------------------------------

// Runnable to clear actors without reporting that the ReadStream has
// closed.  Since this can trigger actor destruction, we need to do
// it on the thread associated with the PBackground actor.  Must be
// cancelable to execute on Worker threads (which can occur when the
// ReadStream is constructed on a child process Worker thread).
class ReadStream::Inner::ForgetRunnable final : public CancelableRunnable {
 public:
  explicit ForgetRunnable(SafeRefPtr<ReadStream::Inner> aStream)
      : CancelableRunnable("dom::cache::ReadStream::Inner::ForgetRunnable"),
        mStream(std::move(aStream)) {}

  NS_IMETHOD Run() override {
    mStream->ForgetOnOwningThread();
    return NS_OK;
  }

  // Note, we must proceed with the Run() method so that we properly
  // call RemoveListener on the actor.
  nsresult Cancel() override {
    Run();
    return NS_OK;
  }

 private:
  ~ForgetRunnable() = default;

  const SafeRefPtr<ReadStream::Inner> mStream;
};

// ----------------------------------------------------------------------------

ReadStream::Inner::Inner(StreamControl* aControl, const nsID& aId,
                         nsIInputStream* aStream)
    : mControl(aControl),
      mId(aId),
      mOwningEventTarget(GetCurrentSerialEventTarget()),
      mState(Open),
      mHasEverBeenRead(false),
      mAsyncOpenStarted(false),
      mMutex("dom::cache::ReadStream"),
      mCondVar(mMutex, "dom::cache::ReadStream"),
      mStream(aStream),
      mSnappyStream(aStream ? new SnappyUncompressInputStream(aStream)
                            : nullptr) {
  MOZ_DIAGNOSTIC_ASSERT(mControl);
  mControl->AddReadStream(SafeRefPtrFromThis());
}

void ReadStream::Inner::Serialize(Maybe<CacheReadStream>* aReadStreamOut,
                                  ErrorResult& aRv) {
  MOZ_ASSERT(mOwningEventTarget->IsOnCurrentThread());
  MOZ_DIAGNOSTIC_ASSERT(aReadStreamOut);
  aReadStreamOut->emplace(CacheReadStream());
  Serialize(&aReadStreamOut->ref(), aRv);
}

void ReadStream::Inner::Serialize(CacheReadStream* aReadStreamOut,
                                  ErrorResult& aRv) {
  MOZ_ASSERT(mOwningEventTarget->IsOnCurrentThread());
  MOZ_DIAGNOSTIC_ASSERT(aReadStreamOut);

  if (mState != Open) {
    aRv.ThrowTypeError(
        "Response body is a cache file stream that has already been closed.");
    return;
  }

  MOZ_DIAGNOSTIC_ASSERT(mControl);

  aReadStreamOut->id() = mId;
  mControl->SerializeControl(aReadStreamOut);

  {
    MutexAutoLock lock(mMutex);
    mControl->SerializeStream(aReadStreamOut, mStream);
  }

  MOZ_DIAGNOSTIC_ASSERT(aReadStreamOut->stream().isNothing() ||
                        aReadStreamOut->stream().ref().stream().type() !=
                            mozilla::ipc::InputStreamParams::T__None);

  // We're passing ownership across the IPC barrier with the control, so
  // do not signal that the stream is closed here.
  Forget();
}

void ReadStream::Inner::CloseStream() {
  MOZ_ASSERT(mOwningEventTarget->IsOnCurrentThread());
  MOZ_ALWAYS_SUCCEEDS(Close());
}

void ReadStream::Inner::CloseStreamWithoutReporting() {
  MOZ_ASSERT(mOwningEventTarget->IsOnCurrentThread());
  Forget();
}

bool ReadStream::Inner::HasEverBeenRead() const {
  MOZ_ASSERT(mOwningEventTarget->IsOnCurrentThread());
  return mHasEverBeenRead;
}

nsresult ReadStream::Inner::Close() {
  // stream ops can happen on any thread
  nsresult rv = NS_OK;
  {
    MutexAutoLock lock(mMutex);
    if (mSnappyStream) {
      rv = mSnappyStream->Close();
    }
  }
  NoteClosed();
  return rv;
}

nsresult ReadStream::Inner::Available(uint64_t* aNumAvailableOut) {
  // stream ops can happen on any thread
  nsresult rv = NS_OK;
  {
    MutexAutoLock lock(mMutex);
    rv = EnsureStream()->Available(aNumAvailableOut);
  }

  if (NS_FAILED(rv)) {
    Close();
  }

  return rv;
}

nsresult ReadStream::Inner::StreamStatus() {
  // stream ops can happen on any thread
  nsresult rv = NS_OK;
  {
    MutexAutoLock lock(mMutex);
    rv = EnsureStream()->StreamStatus();
  }

  if (NS_FAILED(rv)) {
    Close();
  }

  return rv;
}

nsresult ReadStream::Inner::Read(char* aBuf, uint32_t aCount,
                                 uint32_t* aNumReadOut) {
  // stream ops can happen on any thread
  MOZ_DIAGNOSTIC_ASSERT(aNumReadOut);

  nsresult rv = NS_OK;
  {
    MutexAutoLock lock(mMutex);
    rv = EnsureStream()->Read(aBuf, aCount, aNumReadOut);
  }

  if ((NS_FAILED(rv) && rv != NS_BASE_STREAM_WOULD_BLOCK) ||
      *aNumReadOut == 0) {
    Close();
  }

  mHasEverBeenRead = true;

  return rv;
}

nsresult ReadStream::Inner::ReadSegments(nsWriteSegmentFun aWriter,
                                         void* aClosure, uint32_t aCount,
                                         uint32_t* aNumReadOut) {
  // stream ops can happen on any thread
  MOZ_DIAGNOSTIC_ASSERT(aNumReadOut);

  if (aCount) {
    mHasEverBeenRead = true;
  }

  nsresult rv = NS_OK;
  {
    MutexAutoLock lock(mMutex);
    rv = EnsureStream()->ReadSegments(aWriter, aClosure, aCount, aNumReadOut);
  }

  if ((NS_FAILED(rv) && rv != NS_BASE_STREAM_WOULD_BLOCK &&
       rv != NS_ERROR_NOT_IMPLEMENTED) ||
      *aNumReadOut == 0) {
    Close();
  }

  // Verify bytes were actually read before marking as being ever read.  For
  // example, code can test if the stream supports ReadSegments() by calling
  // this method with a dummy callback which doesn't read anything.  We don't
  // want to trigger on that.
  if (*aNumReadOut) {
    mHasEverBeenRead = true;
  }

  return rv;
}

nsresult ReadStream::Inner::IsNonBlocking(bool* aNonBlockingOut) {
  // stream ops can happen on any thread
  MutexAutoLock lock(mMutex);
  if (mSnappyStream) {
    return mSnappyStream->IsNonBlocking(aNonBlockingOut);
  }
  *aNonBlockingOut = false;
  return NS_OK;
}

ReadStream::Inner::~Inner() {
  // Any thread
  MOZ_DIAGNOSTIC_ASSERT(mState == Closed);
  MOZ_DIAGNOSTIC_ASSERT(!mControl);
}

void ReadStream::Inner::NoteClosed() {
  // Any thread
  if (mState == Closed) {
    return;
  }

  if (mOwningEventTarget->IsOnCurrentThread()) {
    NoteClosedOnOwningThread();
    return;
  }

  nsCOMPtr<nsIRunnable> runnable = new NoteClosedRunnable(SafeRefPtrFromThis());
  MOZ_ALWAYS_SUCCEEDS(mOwningEventTarget->Dispatch(runnable.forget(),
                                                   nsIThread::DISPATCH_NORMAL));
}

void ReadStream::Inner::Forget() {
  // Any thread
  if (mState == Closed) {
    return;
  }

  if (mOwningEventTarget->IsOnCurrentThread()) {
    ForgetOnOwningThread();
    return;
  }

  nsCOMPtr<nsIRunnable> runnable = new ForgetRunnable(SafeRefPtrFromThis());
  MOZ_ALWAYS_SUCCEEDS(mOwningEventTarget->Dispatch(runnable.forget(),
                                                   nsIThread::DISPATCH_NORMAL));
}

void ReadStream::Inner::NoteClosedOnOwningThread() {
  MOZ_ASSERT(mOwningEventTarget->IsOnCurrentThread());

  // Mark closed and do nothing if we were already closed
  if (!mState.compareExchange(Open, Closed)) {
    return;
  }

  MaybeAbortAsyncOpenStream();

  MOZ_DIAGNOSTIC_ASSERT(mControl);
  mControl->NoteClosed(SafeRefPtrFromThis(), mId);
  mControl = nullptr;
}

void ReadStream::Inner::ForgetOnOwningThread() {
  MOZ_ASSERT(mOwningEventTarget->IsOnCurrentThread());

  // Mark closed and do nothing if we were already closed
  if (!mState.compareExchange(Open, Closed)) {
    return;
  }

  MaybeAbortAsyncOpenStream();

  MOZ_DIAGNOSTIC_ASSERT(mControl);
  mControl->ForgetReadStream(SafeRefPtrFromThis());
  mControl = nullptr;
}

nsIInputStream* ReadStream::Inner::EnsureStream() {
  mMutex.AssertCurrentThreadOwns();

  // We need to block the current thread while we open the stream.  We
  // cannot do this safely from the main owning thread since it would
  // trigger deadlock.  This should be ok, though, since a blocking
  // stream like this should never be read on the owning thread anyway.
  if (mOwningEventTarget->IsOnCurrentThread()) {
    MOZ_CRASH("Blocking read on the js/ipc owning thread!");
  }

  if (mSnappyStream) {
    return mSnappyStream;
  }

  nsCOMPtr<nsIRunnable> r = NewCancelableRunnableMethod(
      "ReadStream::Inner::AsyncOpenStreamOnOwningThread", this,
      &ReadStream::Inner::AsyncOpenStreamOnOwningThread);
  nsresult rv =
      mOwningEventTarget->Dispatch(r.forget(), nsIThread::DISPATCH_NORMAL);
  if (NS_WARN_IF(NS_FAILED(rv))) {
    OpenStreamFailed();
    return mSnappyStream;
  }

  mCondVar.Wait();
  MOZ_DIAGNOSTIC_ASSERT(mSnappyStream);

  return mSnappyStream;
}

void ReadStream::Inner::AsyncOpenStreamOnOwningThread() {
  MOZ_ASSERT(mOwningEventTarget->IsOnCurrentThread());

  if (mSnappyStream) {
    // Different threads might request opening the stream at the same time. If
    // the earlier request succeeded, then use the result.
    mCondVar.NotifyAll();
    return;
  }

  if (!mControl || mState == Closed) {
    MutexAutoLock lock(mMutex);
    OpenStreamFailed();
    mCondVar.NotifyAll();
    return;
  }

  if (mAsyncOpenStarted) {
    return;
  }
  mAsyncOpenStarted = true;

  RefPtr<ReadStream::Inner> self = this;
  mControl->OpenStream(mId, [self](nsCOMPtr<nsIInputStream>&& aStream) {
    MutexAutoLock lock(self->mMutex);
    self->mAsyncOpenStarted = false;
    if (!self->mStream) {
      if (!aStream) {
        self->OpenStreamFailed();
      } else {
        self->mStream = std::move(aStream);
        self->mSnappyStream = new SnappyUncompressInputStream(self->mStream);
      }
    }
    self->mCondVar.NotifyAll();
  });
}

void ReadStream::Inner::MaybeAbortAsyncOpenStream() {
  if (!mAsyncOpenStarted) {
    return;
  }

  MutexAutoLock lock(mMutex);
  OpenStreamFailed();
  mCondVar.NotifyAll();
}

void ReadStream::Inner::OpenStreamFailed() {
  MOZ_DIAGNOSTIC_ASSERT(!mStream);
  MOZ_DIAGNOSTIC_ASSERT(!mSnappyStream);
  mMutex.AssertCurrentThreadOwns();
  Unused << NS_NewCStringInputStream(getter_AddRefs(mStream), ""_ns);
  mSnappyStream = mStream;
  mStream->Close();
  NoteClosed();
}

// ----------------------------------------------------------------------------

NS_IMPL_ISUPPORTS(cache::ReadStream, nsIInputStream, ReadStream);

// static
already_AddRefed<ReadStream> ReadStream::Create(
    const Maybe<CacheReadStream>& aMaybeReadStream) {
  if (aMaybeReadStream.isNothing()) {
    return nullptr;
  }

  return Create(aMaybeReadStream.ref());
}

// static
already_AddRefed<ReadStream> ReadStream::Create(
    const CacheReadStream& aReadStream) {
  // The parameter may or may not be for a Cache created stream.  The way we
  // tell is by looking at the stream control actor.  If the actor exists,
  // then we know the Cache created it.
  if (!aReadStream.control()) {
    return nullptr;
  }

  MOZ_DIAGNOSTIC_ASSERT(aReadStream.stream().isNothing() ||
                        aReadStream.stream().ref().stream().type() !=
                            mozilla::ipc::InputStreamParams::T__None);

  // Control is guaranteed to survive this method as ActorDestroy() cannot
  // run on this thread until we complete.
  StreamControl* control;
  if (aReadStream.control().IsChild()) {
    auto actor =
        static_cast<CacheStreamControlChild*>(aReadStream.control().AsChild());
    control = actor;
  } else {
    auto actor = static_cast<CacheStreamControlParent*>(
        aReadStream.control().AsParent());
    control = actor;
  }
  MOZ_DIAGNOSTIC_ASSERT(control);

  nsCOMPtr<nsIInputStream> stream = DeserializeIPCStream(aReadStream.stream());

  // Currently we expect all cache read streams to be blocking file streams.
#if defined(MOZ_DIAGNOSTIC_ASSERT_ENABLED)
  if (stream) {
    nsCOMPtr<nsIAsyncInputStream> asyncStream = do_QueryInterface(stream);
    MOZ_DIAGNOSTIC_ASSERT(!asyncStream);
  }
#endif

  return MakeAndAddRef<ReadStream>(MakeSafeRefPtr<ReadStream::Inner>(
      std::move(control), aReadStream.id(), stream));
}

// static
already_AddRefed<ReadStream> ReadStream::Create(
    PCacheStreamControlParent* aControl, const nsID& aId,
    nsIInputStream* aStream) {
  MOZ_DIAGNOSTIC_ASSERT(aControl);

  return MakeAndAddRef<ReadStream>(MakeSafeRefPtr<ReadStream::Inner>(
      static_cast<CacheStreamControlParent*>(aControl), aId, aStream));
}

void ReadStream::Serialize(Maybe<CacheReadStream>* aReadStreamOut,
                           ErrorResult& aRv) {
  mInner->Serialize(aReadStreamOut, aRv);
}

void ReadStream::Serialize(CacheReadStream* aReadStreamOut, ErrorResult& aRv) {
  mInner->Serialize(aReadStreamOut, aRv);
}

ReadStream::ReadStream(SafeRefPtr<ReadStream::Inner> aInner)
    : mInner(std::move(aInner)) {
  MOZ_DIAGNOSTIC_ASSERT(mInner);
}

ReadStream::~ReadStream() {
  // Explicitly close the inner stream so that it does not have to
  // deal with implicitly closing at destruction time.
  mInner->Close();
}

NS_IMETHODIMP
ReadStream::Close() { return mInner->Close(); }

NS_IMETHODIMP
ReadStream::Available(uint64_t* aNumAvailableOut) {
  return mInner->Available(aNumAvailableOut);
}

NS_IMETHODIMP
ReadStream::StreamStatus() { return mInner->StreamStatus(); }

NS_IMETHODIMP
ReadStream::Read(char* aBuf, uint32_t aCount, uint32_t* aNumReadOut) {
  return mInner->Read(aBuf, aCount, aNumReadOut);
}

NS_IMETHODIMP
ReadStream::ReadSegments(nsWriteSegmentFun aWriter, void* aClosure,
                         uint32_t aCount, uint32_t* aNumReadOut) {
  return mInner->ReadSegments(aWriter, aClosure, aCount, aNumReadOut);
}

NS_IMETHODIMP
ReadStream::IsNonBlocking(bool* aNonBlockingOut) {
  return mInner->IsNonBlocking(aNonBlockingOut);
}

}  // namespace mozilla::dom::cache