/**
 * Copyright (c) 2024 Sean DuBois <sean@siobud.com>
 *
 * 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 https://mozilla.org/MPL/2.0/.
 */

#if RTC_ENABLE_MEDIA

#include <algorithm>
#include <memory>

#include "pacinghandler.hpp"

#include "impl/internals.hpp"
#include "impl/threadpool.hpp"

namespace rtc {

PacingHandler::PacingHandler(double bitsPerSecond, std::chrono::milliseconds sendInterval)
    : mBytesPerSecond(bitsPerSecond / 8), mBudget(0.), mSendInterval(sendInterval){};

void PacingHandler::setBitrate(double bitsPerSecond) {
	mBytesPerSecond.store(bitsPerSecond / 8);
}

size_t PacingHandler::clear() {
	std::lock_guard<std::mutex> lock(mMutex);
	const size_t count = mRtpBuffer.size();
	while (!mRtpBuffer.empty())
		mRtpBuffer.pop();
	mQueuedBytes = 0;
	mBudget = 0.;
	return count;
}

void PacingHandler::getQueueStats(size_t &packetCount, size_t &byteCount) {
	std::lock_guard<std::mutex> lock(mMutex);
	packetCount = mRtpBuffer.size();
	byteCount = mQueuedBytes;
}

void PacingHandler::schedule(const message_callback &send) {
	if (mHaveScheduled.exchange(true)) {
		return;
	}

	impl::ThreadPool::Instance().schedule(mSendInterval, [this, weak_this = weak_from_this(),
	                                                      send]() {
		if (auto locked = weak_this.lock()) {
			const std::lock_guard<std::mutex> lock(mMutex);
			mHaveScheduled.store(false);

			// Keep credit for short scheduler stalls. Capping at one interval
			// permanently under-runs the configured bitrate whenever the worker
			// wakes up late, which turns scheduler jitter into growing media delay.
			const double bytesPerSecond = mBytesPerSecond.load();
			const auto now = std::chrono::high_resolution_clock::now();
			auto newBudget =
			    std::chrono::duration<double>(now - mLastRun)
			        .count() *
			    bytesPerSecond;
			const auto catchUpWindow = std::max(mSendInterval * 5,
			                                    std::chrono::milliseconds(25));
			auto maxBudget =
			    std::chrono::duration<double>(catchUpWindow).count() * bytesPerSecond;
			mBudget = std::min(mBudget + newBudget, maxBudget);
			mLastRun = now;

			// Send packets while there is budget, allow a single partial packet over budget
			while (!mRtpBuffer.empty() && mBudget > 0) {
				auto size = int(mRtpBuffer.front()->size());
				send(std::move(mRtpBuffer.front()));
				mRtpBuffer.pop();
				mQueuedBytes = mQueuedBytes >= static_cast<size_t>(size)
				                   ? mQueuedBytes - static_cast<size_t>(size)
				                   : 0;
				mBudget -= size;
			}

			if (!mRtpBuffer.empty()) {
				schedule(send);
			}
		}
	});
}

void PacingHandler::outgoing(message_vector &messages, const message_callback &send) {

	std::lock_guard<std::mutex> lock(mMutex);

	for (auto &m : messages) {
		mQueuedBytes += m->size();
		mRtpBuffer.push(std::move(m));
	}
	messages.clear();

	schedule(send);
}

} // namespace rtc

#endif /* RTC_ENABLE_MEDIA */
