#include "Telemetry/TelemetrySink.h"
#include "SaltyCore.h"
#include "HAL/PlatformFileManager.h"
#include "HAL/RunnableThread.h"
#include "HAL/Event.h"
#include "GenericPlatform/GenericPlatformFile.h"
#include "Misc/CoreDelegates.h"
#include "Misc/Paths.h"

void FLogTelemetrySink::Emit(const FTelemetryEnvelope& Envelope, const FTelemetryEvent& Event)
{
	UE_LOG(LogTelemetry, Log, TEXT("%s"), *Telemetry::ToJsonLine(Envelope, Event));
}

FJsonlTelemetrySink::FJsonlTelemetrySink(const FString& InFilePath, double InDrainIntervalSeconds)
	: FilePath(InFilePath)
	, DrainIntervalSeconds(FMath::Max(InDrainIntervalSeconds, 0.01))
{
	WakeEvent = FPlatformProcess::GetSynchEventFromPool(false);
	DrainedEvent = FPlatformProcess::GetSynchEventFromPool(false);
	Thread = FRunnableThread::Create(this, TEXT("TelemetryJsonlSink"), 0, TPri_BelowNormal);

	OnExitHandle = FCoreDelegates::OnExit.AddRaw(this, &FJsonlTelemetrySink::Flush);
	OnSystemErrorHandle = FCoreDelegates::OnHandleSystemError.AddRaw(this, &FJsonlTelemetrySink::Flush);
}

FJsonlTelemetrySink::~FJsonlTelemetrySink()
{
	FCoreDelegates::OnExit.Remove(OnExitHandle);
	FCoreDelegates::OnHandleSystemError.Remove(OnSystemErrorHandle);

	bStopping = true;
	WakeEvent->Trigger();
	if (Thread)
	{
		Thread->WaitForCompletion();   // Run drains once more on its way out
		delete Thread;
		Thread = nullptr;
	}
	Drain();                           // anything queued after the thread's last pass
	File.Reset();

	FPlatformProcess::ReturnSynchEventToPool(WakeEvent);
	FPlatformProcess::ReturnSynchEventToPool(DrainedEvent);
}

void FJsonlTelemetrySink::Emit(const FTelemetryEnvelope& Envelope, const FTelemetryEvent& Event)
{
	// The whole cost on the caller's thread: one string and one enqueue.
	Queue.Enqueue(Telemetry::ToJsonLine(Envelope, Event));
}

void FJsonlTelemetrySink::Flush()
{
	if (!Thread || bStopping)
	{
		Drain();
		return;
	}

	const uint64 Ticket = ++FlushesRequested;
	WakeEvent->Trigger();
	// Bounded wait: a flush from the crash handler must not hang the process on a wedged disk.
	const double Deadline = FPlatformTime::Seconds() + 5.0;
	while (FlushesCompleted.Load() < Ticket && FPlatformTime::Seconds() < Deadline)
	{
		DrainedEvent->Wait(50);
	}
}

uint32 FJsonlTelemetrySink::Run()
{
	const uint32 IntervalMs = static_cast<uint32>(FMath::Clamp(DrainIntervalSeconds * 1000.0, 10.0, 3600.0 * 1000.0));
	while (!bStopping)
	{
		WakeEvent->Wait(IntervalMs);
		const uint64 Ticket = FlushesRequested.Load();
		Drain();
		FlushesCompleted.Store(Ticket);
		DrainedEvent->Trigger();
	}
	Drain();
	return 0;
}

void FJsonlTelemetrySink::Stop()
{
	bStopping = true;
	WakeEvent->Trigger();
}

void FJsonlTelemetrySink::Drain()
{
	// Single consumer: the worker while it runs, the destructor after it has joined. Never both.
	FString Line;
	bool bWroteAnything = false;
	while (Queue.Dequeue(Line))
	{
		if (!File)
		{
			IPlatformFile& PlatformFile = FPlatformFileManager::Get().GetPlatformFile();
			PlatformFile.CreateDirectoryTree(*FPaths::GetPath(FilePath));
			File.Reset(PlatformFile.OpenWrite(*FilePath, /*bAppend*/ true, /*bAllowRead*/ true));
			if (!File)
			{
				UE_LOG(LogTelemetry, Warning, TEXT("Cannot open telemetry file %s; dropping events"), *FilePath);
				continue;
			}
		}
		Line.AppendChar(TEXT('\n'));
		const FTCHARToUTF8 Utf8(*Line);
		File->Write(reinterpret_cast<const uint8*>(Utf8.Get()), Utf8.Length());
		bWroteAnything = true;
	}
	if (File && bWroteAnything)
	{
		File->Flush();
	}
}