#include "PullMsgAutoDeleteTask.h"
#include "message/message_dto.pb.h"
#include "Message/Tool/MessageTool.h"
#include "Session/Tool/SessionTool.h"
#include "Tools/CsdkCommonTools.h"
#include "Manager/CsdkSubsystem.h"
#include "Tools/CsdkCommonHttp.h"
#include "Common/CsdkErrorCode.h"
#include "Async/TaskGraphInterfaces.h"

using namespace heroim::client::v3beta::dto::message::model;
using namespace heroim::client::v3beta::dto::message;


const FString FPullMsgAutoDeleteTask::Url =TEXT("/spi/v3beta/message/history");

FPullMsgAutoDeleteTask::FPullMsgAutoDeleteTask(const FCsdkSessionId& InSessionId,
	const FString& InAnchorMsgId, const int32 InDesiredPullCount, const ECsdkMsgPullDirection InDirection,
	const TFunction<void(const int32 ErrorCode, const FString& ErrorMsg, const TArray<UCsdkMessageBase*>& Msgs)>&
	InCallback,const TWeakPtr<FCsdkMessageRingBuffer>& InWeakMessageRingBuffer)
{
	SessionId = InSessionId;
	AnchorMsgId = InAnchorMsgId;
	DesiredPullCount = InDesiredPullCount;
	Direction = InDirection;
	Callback =InCallback;
	WeakMessageRingBuffer = InWeakMessageRingBuffer;
}

void FPullMsgAutoDeleteTask::AddReferencedObjects(FReferenceCollector& Collector)
{
	Collector.AddReferencedObjects(PulledAllMessages);
	Collector.AddReferencedObjects(PulledFilteredMessages);
	Collector.AddReferencedObjects(LocalCachedMessages);
}


void FPullMsgAutoDeleteTask::StartTask()
{
	//AnchorMsgId为空并且查下拉取报错，并返回空数组
	if (AnchorMsgId.IsEmpty() && ECsdkMsgPullDirection::Backward ==Direction)
	{
		return AsyncErrorQuit(ECsdkErrorCode::ClientError,TEXT("Can't pull backward By Empty AnchorMsgId"));
	}

	//内存缓存不存在了
	if (!WeakMessageRingBuffer.IsValid())
	{
		return AsyncErrorQuit(ECsdkErrorCode::ClientError,TEXT("local cache is not exist"));
	}
	const TSharedPtr<FCsdkMessageRingBuffer> RingBuffer = WeakMessageRingBuffer.Pin();
	LocalCachedMessages = RingBuffer->GetFilteredMsgs(AnchorMsgId, DesiredPullCount, Direction);
	
	//内存缓存够用
	if (LocalCachedMessages.Num() >= DesiredPullCount ||!RingBuffer->GetRemoteHeadHasMore())
	{
		AsyncTask(ENamedThreads::GameThread, [this]()
		{
			Completed(TEXT("Local Cached Msg"));
		});
		return;
	}

	//内存不够，需要从远程拉取
	if(LocalCachedMessages.Num()>0)
	{
		if(ECsdkMsgPullDirection::Forward ==  Direction)
		{
			StartHttpRequest(MakeRequest(LocalCachedMessages[0]->MsgId));
		}
		else
		{
			StartHttpRequest(MakeRequest(LocalCachedMessages.Last()->MsgId));
		}
	}
	else
	{
		StartHttpRequest(MakeRequest(AnchorMsgId));
	}

}

MessageHistoryRequest FPullMsgAutoDeleteTask::MakeRequest(const FString& InAnchorMsgId)
{
	const TSharedPtr<FCsdkMessageRingBuffer> RingBuffer = WeakMessageRingBuffer.Pin();
	MessageHistoryRequest Request;
	Request.set_allocated_base(FCsdkCommonTools::MakePbBase());
	Request.set_allocated_sessionkey(CsdkSessionTool::MakePbSessionKey(SessionId));
	Request.set_size(DesiredPullCount);
	Request.set_lastmsgid(FSTRING_TO_STD_STRING(InAnchorMsgId));
	Request.set_direction(static_cast<QueryDirection>(Direction));
	return Request;
}

void FPullMsgAutoDeleteTask::StartHttpRequest(const MessageHistoryRequest& Request)
{
	const auto HttpRequest = FHttpModule::Get().CreateRequest();
	HttpRequest->SetURL(FCsdkCommonTools::GetHttpAddress()+ Url);
	HttpRequest->SetVerb(TEXT("POST"));
	FCsdkCommonTools::HttpV2SetDefaultHeader(HttpRequest);
	HttpRequest->SetContent(FCsdkCommonTools::ConvertPbStructToBytes(Request));
	HttpRequest->OnProcessRequestComplete().BindRaw(this,&FPullMsgAutoDeleteTask::OnHttpResponse);
	HttpRequest->ProcessRequest();
	const FString DirectionStr =StaticEnum<ECsdkMsgPullDirection>()->GetNameStringByValue(static_cast<int64>( Direction));
	CSDK_LOG(LogChatSdk,Log,TEXT("Start Request:%s,SessionId:%s,DesirePullCount:%d,Direction:%s"),*Url,*SessionId.ToString(),DesiredPullCount,*DirectionStr);
}

void FPullMsgAutoDeleteTask::OnHttpResponse(FHttpRequestPtr Request, FHttpResponsePtr Response, bool bConnectedSuccessfully)
{
	//当前消息缓存已经不存在了
	if(!WeakMessageRingBuffer.IsValid())
	{
		ErrorQuit(ECsdkErrorCode::ClientError,TEXT("MessageCache Is Destroy while Http Response"));
		return;
	}
	
	if(!bConnectedSuccessfully) //连接不成功，就不返回数据
	{
		ErrorQuit(ECsdkErrorCode::UnReachable,TEXT("Request Failed, can't connect to Sever"));
		return;
	}
	
	const TApiResult<MessageHistoryResponse> Result =TApiResult<MessageHistoryResponse>::ParseAPIResult(Response->GetContent());
	
	CSDK_LOG(LogChatSdk,Log,TEXT("Url: %s,debugMsg: %s, SessionId:%s"),*Url,*Result.FormatDebugMsg(),*SessionId.ToString());
	
	if(Result.Code!=ECsdkErrorCode::Success) //服务器返回错误，就不返回数据
	{
		ErrorQuit(Result.Code,Result.Msg);
		return;
	}
	if (SessionId.Type == ECsdkSessionType::PrivateChat)
	{
		FCsdkCommonTools::GetSessionManager().UpdatePrivateChatSequence(SessionId,Result.EmbedData.readseq());
	}


	FillData(Result.EmbedData.list());
	
	//拉取数据小于需求数量
	if( Result.EmbedData.list_size() < DesiredPullCount )
	{
		Completed(Result.Msg,false);
		return;
	}
	
	//拉取数据足够了
	if(PulledFilteredMessages.Num() + LocalCachedMessages.Num()>= DesiredPullCount)
	{
		Completed(Result.Msg);
		return;
	}
	
	
	if(ECsdkMsgPullDirection::Forward ==  Direction )
	{
		
		StartHttpRequest(MakeRequest(PulledAllMessages[0]->MsgId));
	}
	else
	{
		StartHttpRequest(MakeRequest(PulledAllMessages.Last()->MsgId));
	}
}

void FPullMsgAutoDeleteTask::FillData(const ::herogoogle::protobuf::RepeatedPtrField< ::heroim::client::v3beta::dto::message::model::Message >& List)
{
	TArray<UCsdkMessageBase*> Msgs;//本次拉取的历史消息
	for(const auto& PbMsg : List) //构造消息UObject
	{
		Msgs.Add(MessageTool::NewMsgObj(FCsdkManager::GetInstance().Subsystem,PbMsg));
	}
	
	TArray<UCsdkMessageBase*> Filtered;
	if(SessionId.Type == ECsdkSessionType::PrivateChat)//私聊不屏蔽消息
	{
		Filtered = Msgs;
	}
	else
	{
		Filtered = FCsdkCommonTools::GetBlockManager().FilterMsgsSenderNotBlocked(Msgs);
	}
	
	//填充拉取消息
	if(ECsdkMsgPullDirection::Forward ==  Direction )
	{
		TArray<UCsdkMessageBase*> Temp;
		Temp.Append(Msgs);
		Temp.Append(PulledAllMessages);
		PulledAllMessages = Temp;
		
		Filtered.Append(PulledFilteredMessages);
		PulledFilteredMessages = Filtered;
	}
	else
	{
		PulledAllMessages.Append(Msgs);
		PulledFilteredMessages.Append(Filtered);
	}
}


void FPullMsgAutoDeleteTask::TryAddToRingBuffer(const TArray<UCsdkMessageBase*>& Msgs,const bool bIsFullCount) const
{
	if (ECsdkMsgPullDirection::Forward != Direction)
		return;
		
	const TSharedPtr<FCsdkMessageRingBuffer> RingBuffer = WeakMessageRingBuffer.Pin();

	for (int32 i = Msgs.Num() - 1; i >= 0; i--) //填入缓存
	{
		RingBuffer->TryPushFront(Msgs[i]);
	}
	if(!bIsFullCount && !RingBuffer->IsFull())//拉取消息不足，且缓存不满
	{
		RingBuffer->SetRemoteHeadHasMore(false);
	}
}

TArray<TObjectPtr<UCsdkMessageBase>> FPullMsgAutoDeleteTask::AssembleResult() const
{
	TArray<TObjectPtr<UCsdkMessageBase>> Result;
	if (ECsdkMsgPullDirection::Forward == Direction)
	{
		TArray<TObjectPtr<UCsdkMessageBase>> Total;
		Total.Append(PulledFilteredMessages);
		Total.Append(LocalCachedMessages);
		for (int32 i = Total.Num() - DesiredPullCount > 0 ? Total.Num() - DesiredPullCount : 0; i < Total.Num(); i++)
		{
			Result.Add(Total[i]);
		}
		
	}
	else
	{
		TArray<TObjectPtr<UCsdkMessageBase>> Total;
		Total.Append(LocalCachedMessages);
		Total.Append(PulledFilteredMessages);

		for (int32 i = 0; i < Total.Num() && i < DesiredPullCount; i++)
		{
			Result.Add(Total[i]);
		}
	}
	return Result;
}

void FPullMsgAutoDeleteTask::Completed(const FString& ErrorMsg,const bool bIsFullCount)
{
	//内存缓存不存在了
	if (!WeakMessageRingBuffer.IsValid())
	{
		CSDK_LOG(LogChatSdk,Error,TEXT("Local Cached Msg RingBuffer Does not Exist"))
		delete this;
		return;
	}

	if(AnchorMsgId.IsEmpty())//如果AnchorMsgId，此时缓存里面可能有新的TCP数据过来了,重新获取一遍缓存数据
	{
		const TSharedPtr<FCsdkMessageRingBuffer> RingBuffer = WeakMessageRingBuffer.Pin();
		LocalCachedMessages = RingBuffer->GetFilteredMsgs(AnchorMsgId, DesiredPullCount, Direction);
	}

	TryAddToRingBuffer(PulledAllMessages,bIsFullCount);

	
	const TArray<UCsdkMessageBase*> Result =AssembleResult();
	
	CSDK_LOG(LogChatSdk,Log,TEXT("Pull Msg Num:%d,Auto Delte self,ErrorMsg:%s"),Result.Num(),*ErrorMsg);
	Callback(ECsdkErrorCode::Success,ErrorMsg,Result);

	delete this;
}


void FPullMsgAutoDeleteTask::AsyncErrorQuit(const ECsdkErrorCode::Type& ErrorCode,const FString& ErrorMsg) const
{
	AsyncTask(ENamedThreads::GameThread, [this,ErrorCode,ErrorMsg]()
	{
		ErrorQuit(ErrorCode,ErrorMsg);
	});
}


void FPullMsgAutoDeleteTask::ErrorQuit(const int32 ErrorCode, const FString& ErrorMsg) const
{
	Callback(ErrorCode,ErrorMsg,{});
	CSDK_LOG(LogChatSdk,Error,TEXT("FPullMsgAutoDeleteTask ErrorQuit: SessionId:%s, ErrorCode:%d, ErrorMsg: %s,Auto Delte self"),*SessionId.ToString(),ErrorCode,*ErrorMsg);
	delete this;
}
