
package org.tio.sitexxx.im.server.handler.wx;

import cn.hutool.core.util.StrUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tio.core.ChannelContext;
import org.tio.jfinal.kit.Ret;
import org.tio.sitexxx.service.model.main.User;
import org.tio.sitexxx.service.model.main.WxChatItemsMeta;
import org.tio.sitexxx.service.model.main.WxGroupMsg;
import org.tio.sitexxx.service.service.chat.ChatMsgService;
import org.tio.sitexxx.service.utils.LockUtils.LockPrefix;
import org.tio.sitexxx.service.utils.RetUtils;
import org.tio.sitexxx.service.vo.*;
import org.tio.sitexxx.service.vo.wx.FocusVo;
import org.tio.utils.json.Json;
import org.tio.utils.lock.LockUtils;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock.WriteLock;

/**
 * wx的队列API
 * @author xufei
 * 2020年1月17日 下午2:25:27
 */
public class WxChatQueueApi {

	private static Logger log = LoggerFactory.getLogger(WxChatQueueApi.class);

	public static final WxChatQueueApi me = new WxChatQueueApi();

	/**
	 * 焦点队列数
	 */
	private static final int focusIndex = 5;

	/**
	 * 焦点队列
	 */
	private static ArrayList<LinkedBlockingQueue<FocusVo>> focusQueue = new ArrayList<>();

	/**
	 * 群发送后处理队列数
	 */
	private static final int groupSendMsgAfterIndex = 10;

	/**
	 * 群发送后处理队列
	 */
	private static ArrayList<LinkedBlockingQueue<AfterGroupMsgSendVo>> groupSendMsgAfterQueue = new ArrayList<>();

	/**
	 * 群处理合并
	 */
	private static ArrayList<Map<String, ChatMsgMergeVo>> groupMergeMsgMap = new ArrayList<>();

	/**
	 * 队列初始化
	 * @author xufei
	 * 2020年10月12日 下午3:17:04
	 */
	public static void wxQueueInit() {
		focusQueueInit();
		groupSendMsgAfterQueueInit();
	}

	/**
	 * 警告：目前分布式使用是在im服务器分布式，http服务器依赖nginx服务分布，此队列分布式请谨慎扩展
	 * 1、如果未来消息存储和已读未读逻辑转存在客户端，此队列可废除
	 * 2、如果未来im分布式算法更改，此方法有可能会引起数据一致性问题和并发同步问题
	 * 3、此队列如果涉及到短链接-http，有可能出现数据一致性问题
	 * 
	 * 初始化队列
	 * 该队列列表进行焦点处理
	 * 1、如果未来一个用户一次会话只在一个服务器上请求，该方法不变
	 * 2、如果未来一个用户一次会话会时刻变动（10个请求有可能出跳到10个不同的服务器），该方法需要进行分布式调整
	 * @author xufei
	 * 2020年6月18日 下午2:54:37
	 */
	public static void focusQueueInit() {
		for (int i = 0; i < focusIndex; i++) {
			final int qindex = i;
			focusQueue.add(new LinkedBlockingQueue<>());
			new Thread(() -> {
				int error = 0;
				while (true) {
					FocusVo focusVo;
					try {
						focusVo = focusQueue.get(qindex).take();
						error = 0;
					} catch (InterruptedException e1) {
						error++;
						log.error("队列获取焦点对象失败，队列index:{}，错误次数：{}", qindex, error, e1);
						if (error >= 5) {
							log.error("队列联系获取焦点对象5次失败，退出循环，队列index:{}", qindex);
							break;
						}
						continue;
					}
					try {
						if (focusVo == null) {
							log.error("焦点对象为空，队列index:{}", qindex);
							continue;
						}
						if (focusVo.getOper() == null) {
							log.error("焦点对象操作类型为空,focusvo:{}", Json.toJson(focusVo));
							continue;
						}
						switch (focusVo.getOper()) {
						case FocusVo.operCode.JOIN:
							Ret ret = ChatMsgService.joinDeal(focusVo);
							Short chatmode = focusVo.getChatmode();
							Integer uid = focusVo.getUid();
							WxChatApi.focusNtf(uid, focusVo.getRchatlinkid());
							if (ret.isFail()) {
								if (Objects.equals(chatmode, Const.ChatMode.GROUP)) {
									WxChatApi.sendFriendErrorMsg(uid, uid, uid, focusVo.getRchatlinkid(),
											AppCode.GroupErrorCode.SYS_ERROR, RetUtils.getRetMsg(ret));
								} else {
									WxChatApi.sendFriendErrorMsg(uid, uid, uid, focusVo.getRchatlinkid(),
											AppCode.FriendErrorCode.SYS_ERROR, RetUtils.getRetMsg(ret));
								}
								continue;
							}
							boolean isToRead = RetUtils.getOkTData(ret);
							if (isToRead) {
								Long tochatlinkid = RetUtils.getOkTData(ret, "tochatlinkid");
								Integer touid = RetUtils.getOkTData(ret, "touid");
								if (WxSynApi.isSynVersion()) {
									if (Objects.equals(chatmode, Const.ChatMode.P2P)) {
										WxSynApi.synMsgFocusJoin(tochatlinkid, touid, focusVo);
									}
								} else {
									if (Objects.equals(chatmode, Const.ChatMode.GROUP)) {
										WxChatApi.userChatOper(focusVo.getUid(),
												-focusVo.getGroupid(), Const.WxUserOper.TO_READ, "你已读群消息");
									} else {
										WxChatApi.userChatOper(touid, tochatlinkid,
												Const.WxUserOper.TO_READ, "好友已读你的消息");
									}
								}

							}
							break;
						case FocusVo.operCode.LEAVE:
							ChatMsgService.leaveDeal(focusVo);
							WxChatApi.focusNtf(focusVo.getUid(), null);
							break;
						case FocusVo.operCode.REFRESH:
							ChatMsgService.refreshDeal(focusVo);
							break;
						default:
							log.error("无效焦点操作：opercode:{}", focusVo.getOper());
							break;
						}
					} catch (Exception e) {
						log.error("", e);
					}
				}
			}, "focusQueueInit").start();
		}
	}

	/**
	 * 后消息逻辑处理
	 * @author xufei
	 * 2020年10月12日 下午2:19:31
	 */
	public static void groupSendMsgAfterQueueInit() {
		for (int i = 0; i < groupSendMsgAfterIndex; i++) {
			final int qindex = i;
			groupSendMsgAfterQueue.add(new LinkedBlockingQueue<>());
			groupMergeMsgMap.add(new HashMap<>());
			new Thread(() -> {
				int error = 0;
				while (true) {
					AfterGroupMsgSendVo afterVo;
					try {
						afterVo = groupSendMsgAfterQueue.get(qindex).take();
						error = 0;
					} catch (InterruptedException e1) {
						error++;
						log.error("队列获取群发送对象失败，队列index:{}，错误次数：{}", qindex, error, e1);
						if (error >= 5) {
							log.error("队列联系获取群发送对象5次失败，退出循环，队列index:{}", qindex);
							break;
						}
						continue;
					}
					try {
						if (afterVo == null) {
							log.error("群消息对象为空，队列index:{}", qindex);
							continue;
						}
						if (afterVo.getMsg() == null) {
							log.error("群消息对象消息为空,focusvo:{}", Json.toJson(afterVo));
							continue;
						}
						Long groupid = afterVo.getMsg().getGroupid();
						ReentrantReadWriteLock rwLock = LockUtils.getReentrantReadWriteLock(LockPrefix.CHAT_CHATITEM_META_KEY_PREFIX + "." + groupid, WxChatItemsMeta.class);
						WriteLock writeLock = rwLock.writeLock();
						writeLock.lock();
						try {
							//未读处理
							ChatMsgService.me.afterSendGroupChatMsg(afterVo.getMsg(), 1, Const.YesOrNo.YES);
							//已读处理
							ChatMsgService.me.afterSendGroupChatMsg(afterVo.getMsg(), null, Const.YesOrNo.YES);
							if (StrUtil.isNotBlank(afterVo.getAts())) {
								//at逻辑处理
								ChatMsgService.me.afterSendGroupAtMsg(afterVo.getMsg(), afterVo.getAts(), 1, Const.YesOrNo.YES);
							}
						} catch (Exception e) {
							// TODO: handle exception
						} finally {
							writeLock.unlock();
						}
					} catch (Exception e) {
						log.error("", e);
					}
				}
			}, "groupSendMsgAfterQueueInit").start();
		}
	}

	/**
	 * 离开焦点添加队列
	 * @param curr
	 * @param devicetype
	 * @author xufei
	 * 2020年6月5日 下午2:54:07
	 */
	public static void leaveFocusQueue(User curr, Short devicetype, String cid) {
		FocusVo focusVo = new FocusVo();
		focusVo.setUid(curr.getId());
		focusVo.setDevicetype(devicetype);
		focusVo.setOper(FocusVo.operCode.LEAVE);
		focusVo.setChannelid(cid);
		int queueindex = curr.getId() % focusIndex;
		try {
			//此处会阻塞队列，如果是短链接谨慎使用
			focusQueue.get(queueindex).put(focusVo);
		} catch (InterruptedException e) {
			log.error("", e);
		}
	}

	/**
	 * 刷新焦点添加队列
	 * @param curr
	 * @param devicetype
	 * @author xufei
	 * 2020年6月5日 下午2:58:06
	 */
	public static void refreshFocusQueue(User curr, Short devicetype, String cid) {
		FocusVo focusVo = new FocusVo();
		focusVo.setUid(curr.getId());
		focusVo.setDevicetype(devicetype);
		focusVo.setOper(FocusVo.operCode.REFRESH);
		focusVo.setChannelid(cid);
		int queueindex = curr.getId() % focusIndex;
		try {
			//此处会阻塞队列，如果是短链接谨慎使用
			focusQueue.get(queueindex).put(focusVo);
		} catch (InterruptedException e) {
			log.error("", e);
		}
	}

	/**
	 * 进入焦点添加队列
	 * @param curr
	 * @param chatlinkid
	 * @param groupid
	 * @param chatmode
	 * @param devicetype
	 * @param ipid
	 * @author xufei
	 * 2020年6月5日 下午2:46:04
	 */
	public static void joinFocusQueue(ChannelContext channelContext, User curr, Long chatlinkid, Long groupid, Short chatmode, Short devicetype, Integer ipid) {
		if (Objects.equals(devicetype, Devicetype.IOS.getValue()) || Objects.equals(devicetype, Devicetype.ANDROID.getValue())) {
			devicetype = Devicetype.APP.getValue();
		}
		FocusVo focusVo = new FocusVo();
		focusVo.setChannelid(channelContext.getId());
		focusVo.setUid(curr.getId());
		focusVo.setChatlinkid(chatlinkid);
		focusVo.setDevicetype(devicetype);
		focusVo.setChatmode(chatmode);
		focusVo.setIpid(ipid);
		focusVo.setGroupid(groupid);
		focusVo.setOper(FocusVo.operCode.JOIN);
		int queueindex = curr.getId() % focusIndex;
		try {
			//此处会阻塞队列，如果是短链接谨慎使用
			focusQueue.get(queueindex).put(focusVo);
		} catch (InterruptedException e) {
			log.error("", e);
		}
	}

	/**
	 * TODO:此处可以进行数据库操作合并-待优化
	 * @param groupid
	 * @param msg
	 * @param notreadcount
	 * @param ats
	 * @author xufei
	 * 2020年10月12日 下午2:14:56
	 */
	public static void joinGroupSendMsgAfterQueue(Long groupid, WxGroupMsg msg, Short notreadcount, String ats) {
		AfterGroupMsgSendVo sendVo = new AfterGroupMsgSendVo(msg, notreadcount, ats);
		long queueindex = groupid % groupSendMsgAfterIndex;
		try {
			//此处会阻塞队列，如果是短链接谨慎使用
			groupSendMsgAfterQueue.get((int) queueindex).put(sendVo);
		} catch (InterruptedException e) {
			log.error("", e);
		}
	}
}
