/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

//#define _DEFAULT_SOURCE
#include <stdint.h>
#include "os.h"
#include "tulog.h"
#include "tglobal.h"
#include "tsocket.h"
#include "trpc.h"
#include "tqueue.h"
#include "twal.h"
#include "tsync.h"

int msgSize = 128;
int commit = 0;
int dataFd = -1;
void *qhandle = NULL;
int walNum = 0;
uint64_t  tversion = 0;
void *syncHandle;
int   role;
int  nodeId;
char path[256];
int  numOfWrites ;
SSyncInfo syncInfo;
SSyncCfg *pCfg;

int writeIntoWal(SWalHead *pHead)
{ 
  if (dataFd < 0) {
    char  walName[280];
    snprintf(walName, sizeof(walName), "%s/wal/wal.%d", path, walNum);
    remove(walName);
    dataFd = open(walName, O_CREAT | O_WRONLY, S_IRWXU | S_IRWXG | S_IRWXO);  
    if (dataFd < 0) { 
      uInfo("failed to open wal file:%s(%s)", walName, strerror(errno));
      return -1;
    } else {
      walNum++;
      uInfo("file:%s is opened to write, walNum:%d", walName, walNum);
    }
  }

  if (write(dataFd, pHead, sizeof(SWalHead) + pHead->len) < 0) {
    uError("ver:%" PRIu64 ", failed to write wal file(%s)", pHead->version, strerror(errno));
  } else {
    uDebug("ver:%" PRIu64 ", written to wal", pHead->version);
  }

  numOfWrites++;
  if (numOfWrites >= 10000) {
    uInfo("%d request have been written into disk", numOfWrites);
    close(dataFd);
    dataFd = -1;
    numOfWrites = 0;
  }
 
  return 0;
}

void confirmForward(void *ahandle, void *mhandle, int32_t code)
{
  SRpcMsg  *pMsg = (SRpcMsg *)mhandle;
  SWalHead *pHead = (SWalHead *)(((char *)pMsg->pCont) - sizeof(SWalHead));

  uDebug("ver:%" PRIu64 ", confirm is received", pHead->version);

  rpcFreeCont(pMsg->pCont);

  SRpcMsg    rpcMsg;
  rpcMsg.pCont = rpcMallocCont(msgSize);
  rpcMsg.contLen = msgSize;
  rpcMsg.handle = pMsg->handle;
  rpcMsg.code = code;
  rpcSendResponse(&rpcMsg);

  taosFreeQitem(mhandle); 
}

int processRpcMsg(void *item) {
  SRpcMsg   *pMsg = (SRpcMsg *)item;
  SWalHead  *pHead = (SWalHead *)(((char *)pMsg->pCont) - sizeof(SWalHead));
  int        code = -1;

  if (role != TAOS_SYNC_ROLE_MASTER) {
    uError("not master, write failed, role:%s", syncRole[role]);
  } else {

    pHead->version = ++tversion;
    pHead->msgType = pMsg->msgType;
    pHead->len = pMsg->contLen;

    uDebug("ver:%" PRIu64 ", pkt from client processed", pHead->version);
    writeIntoWal(pHead); 
    syncForwardToPeer(syncHandle, pHead, item, TAOS_QTYPE_RPC);

    code = 0;
  }

  if (pCfg->quorum <= 1) { 
    taosFreeQitem(item); 
    rpcFreeCont(pMsg->pCont);

    SRpcMsg    rpcMsg;
    rpcMsg.pCont = rpcMallocCont(msgSize);
    rpcMsg.contLen = msgSize;
    rpcMsg.handle = pMsg->handle;
    rpcMsg.code = code;
    rpcSendResponse(&rpcMsg);
  }

  return code;
}

int processFwdMsg(void *item) {

  SWalHead *pHead = (SWalHead *)item;

  if (pHead->version <= tversion) {
    uError("ver:%" PRIu64 ", forward is even lower than local:%" PRIu64, pHead->version, tversion);
    return -1;
  }

  uDebug("ver:%" PRIu64 ", forward from peer is received", pHead->version);
  writeIntoWal(pHead);
  tversion = pHead->version;

  if (pCfg->quorum > 1) syncConfirmForward(syncHandle, pHead->version, 0);

  // write into cache

/*
  if (pHead->handle) {
    syncSendFwdAck(syncHandle, pHead->handle, 0);  
  }
*/

  taosFreeQitem(item);

  return 0;
}

int processWalMsg(void *item) {

  SWalHead *pHead = (SWalHead *)item;

  if (pHead->version <= tversion) {
    uError("ver:%" PRIu64 ", wal is even lower than local:%" PRIu64, pHead->version, tversion);
    return -1;
  };

  uDebug("ver:%" PRIu64 ", wal from peer is received", pHead->version);
  writeIntoWal(pHead);
  tversion = pHead->version;

  // write into cache

/*
  if (pHead->handle) {
    syncSendFwdAck(syncHandle, pHead->handle, 0);  
  }
*/

  taosFreeQitem(item);

  return 0;
}

void *processWriteQueue(void *param) {
  int        type;
  void      *item;

  while (1) {
    int ret = taosReadQitem(qhandle, &type, &item);
    if (ret <= 0) {
      usleep(1000);
      continue;
    }     

    if (type == TAOS_QTYPE_RPC) {
      processRpcMsg(item);
    } else if (type == TAOS_QTYPE_WAL) {
      processWalMsg(item);
    } else if (type == TAOS_QTYPE_FWD) {
      processFwdMsg(item);
    } 

  }

  return NULL;
}

int retrieveAuthInfo(char *meterId, char *spi, char *encrypt, char *secret, char *ckey) {
  // app shall retrieve the auth info based on meterID from DB or a data file
  // demo code here only for simple demo
  int ret = 0;

  if (strcmp(meterId, "michael") == 0) {
    *spi = 1;
    *encrypt = 0;
    strcpy(secret, "mypassword");
    strcpy(ckey, "key");
  } else if (strcmp(meterId, "jeff") == 0) {
    *spi = 0;
    *encrypt = 0;
  } else {
    ret = -1;  // user not there
  }

  return ret;
}

void processRequestMsg(SRpcMsg *pMsg, SRpcEpSet *pEpSet) {
  
  SRpcMsg *pTemp;

  pTemp = taosAllocateQitem(sizeof(SRpcMsg));
  memcpy(pTemp, pMsg, sizeof(SRpcMsg));
  
  uDebug("request is received, type:%d, len:%d", pMsg->msgType, pMsg->contLen);
  taosWriteQitem(qhandle, TAOS_QTYPE_RPC, pTemp); 
}

uint32_t getFileInfo(void *ahandle, char *name, uint32_t *index, uint32_t eindex, int32_t *size, uint64_t *fversion) 
{
  uint32_t     magic;
  struct stat  fstat;
  char         aname[280];

  if (*index == 2) {
    uInfo("wait for a while .....");
    sleep(3);
  }

  if (name[0] == 0) {
    // find the file 
    snprintf(aname, sizeof(aname), "%s/data/data.%d", path, *index);
    sprintf(name, "data/data.%d", *index); 
  } else {
    snprintf(aname, sizeof(aname), "%s/%s", path, name);
  }

  uInfo("get file info:%s", aname);
  if ( stat(aname, &fstat) < 0 ) return 0; 

  *size = fstat.st_size;
  magic = fstat.st_size;

  return magic;
}

int  getWalInfo(void *ahandle, char *name, uint32_t *index) {

  struct stat  fstat;
  char         aname[280];

  name[0] = 0;
  if (*index + 1> walNum) return 0;

  snprintf(aname, sizeof(aname), "%s/wal/wal.%d", path, *index);
  sprintf(name, "wal/wal.%d", *index); 
  uInfo("get wal info:%s", aname);

  if ( stat(aname, &fstat) < 0 ) return -1;

  if (*index >= walNum-1) return 0;  // no more

  return 1;

}

int writeToCache(void *ahandle, void *data, int type) {
  SWalHead *pHead = data;

  uDebug("pkt from peer is received, ver:%" PRIu64 " len:%d type:%d", pHead->version, pHead->len, type);

  int   msgSize = pHead->len + sizeof(SWalHead);
  void *pMsg = taosAllocateQitem(msgSize);
  memcpy(pMsg, pHead, msgSize);
  taosWriteQitem(qhandle, type, pMsg); 

  return 0;
}

void confirmFwd(void *ahandle, int64_t version) {

  return;
}

void notifyRole(void *ahandle, int8_t r) {
  role = r;
  printf("current role:%s\n", syncRole[role]);
}


void initSync() {

  pCfg->replica = 1;
  pCfg->quorum = 1;
  syncInfo.vgId = 1;
  syncInfo.ahandle = &syncInfo;
  syncInfo.getFileInfo = getFileInfo;
  syncInfo.getWalInfo = getWalInfo;
  syncInfo.writeToCache = writeToCache;
  syncInfo.confirmForward = confirmForward;
  syncInfo.notifyRole = notifyRole;

  pCfg->nodeInfo[0].nodeId = 1;
  pCfg->nodeInfo[0].nodePort = 7010;
  taosGetFqdn(pCfg->nodeInfo[0].nodeFqdn);

  pCfg->nodeInfo[1].nodeId = 2;
  pCfg->nodeInfo[1].nodePort = 7110;
  taosGetFqdn(pCfg->nodeInfo[1].nodeFqdn);

  pCfg->nodeInfo[2].nodeId = 3;
  pCfg->nodeInfo[2].nodePort = 7210;
  taosGetFqdn(pCfg->nodeInfo[2].nodeFqdn);

  pCfg->nodeInfo[3].nodeId = 4;
  pCfg->nodeInfo[3].nodePort = 7310;
  taosGetFqdn(pCfg->nodeInfo[3].nodeFqdn);

  pCfg->nodeInfo[4].nodeId = 5;
  pCfg->nodeInfo[4].nodePort = 7410;
  taosGetFqdn(pCfg->nodeInfo[4].nodeFqdn);
}

void doSync()
{
  for (int i=0; i<5; ++i) {
    if (tsSyncPort == pCfg->nodeInfo[i].nodePort) 
      nodeId = pCfg->nodeInfo[i].nodeId;
  }

  snprintf(path, sizeof(path), "/root/test/d%d", nodeId);
  strcpy(syncInfo.path, path);

  if ( syncHandle == NULL) {
      syncHandle = syncStart(&syncInfo);
  } else {
      if (syncReconfig(syncHandle, pCfg) < 0) syncHandle = NULL;
  }

  uInfo("nodeId:%d path:%s syncPort:%d", nodeId, path, tsSyncPort);
}

int main(int argc, char *argv[]) {
  SRpcInit rpcInit;
  char     dataName[20] = "server.data";
  pCfg =  &syncInfo.syncCfg;

  initSync();

  memset(&rpcInit, 0, sizeof(rpcInit));
  rpcInit.localPort    = 7000;
  rpcInit.label        = "SER";
  rpcInit.numOfThreads = 1;
  rpcInit.cfp          = processRequestMsg;
  rpcInit.sessions     = 1000;
  rpcInit.idleTime     = tsShellActivityTimer*1500; 
  rpcInit.afp          = retrieveAuthInfo;

  for (int i=1; i<argc; ++i) {
    if (strcmp(argv[i], "-p")==0 && i < argc-1) {
      rpcInit.localPort = atoi(argv[++i]);
    } else if (strcmp(argv[i], "-t")==0 && i < argc-1) {
      rpcInit.numOfThreads = atoi(argv[++i]);
    } else if (strcmp(argv[i], "-m")==0 && i < argc-1) {
      msgSize = atoi(argv[++i]);
    } else if (strcmp(argv[i], "-s")==0 && i < argc-1) {
      rpcInit.sessions = atoi(argv[++i]);
    } else if (strcmp(argv[i], "-o")==0 && i < argc-1) {
      tsCompressMsgSize = atoi(argv[++i]);
    } else if (strcmp(argv[i], "-w")==0 && i < argc-1) {
      commit = atoi(argv[++i]);
    } else if (strcmp(argv[i], "-v")==0 && i < argc-1) {
      syncInfo.version = atoi(argv[++i]);
    } else if (strcmp(argv[i], "-r")==0 && i < argc-1) {
      pCfg->replica = atoi(argv[++i]);
    } else if (strcmp(argv[i], "-q")==0 && i < argc-1) {
      pCfg->quorum = atoi(argv[++i]);
    } else if (strcmp(argv[i], "-d")==0 && i < argc-1) {
      rpcDebugFlag = atoi(argv[++i]);
    } else {
      printf("\nusage: %s [options] \n", argv[0]);
      printf("  [-p port]: server port number, default is:%d\n", rpcInit.localPort);
      printf("  [-t threads]: number of rpc threads, default is:%d\n", rpcInit.numOfThreads);
      printf("  [-s sessions]: number of sessions, default is:%d\n", rpcInit.sessions);
      printf("  [-m msgSize]: message body size, default is:%d\n", msgSize);
      printf("  [-o compSize]: compression message size, default is:%d\n", tsCompressMsgSize);
      printf("  [-w write]: write received data to file(0, 1, 2), default is:%d\n", commit);
      printf("  [-v version]: initial node version, default is:%ld\n", syncInfo.version);
      printf("  [-r replica]: replicacation number, default is:%d\n", pCfg->replica);
      printf("  [-q quorum]: quorum, default is:%d\n", pCfg->quorum);
      printf("  [-d debugFlag]: debug flag, default:%d\n", rpcDebugFlag);
      printf("  [-h help]: print out this help\n\n");
      exit(0);
    }
  }
 
  uDebugFlag = rpcDebugFlag;
  dDebugFlag = rpcDebugFlag; 
  //tmrDebugFlag = rpcDebugFlag; 
  tsAsyncLog = 0;
  taosInitLog("server.log", 1000000, 10);

  rpcInit.connType = TAOS_CONN_SERVER;
  void *pRpc = rpcOpen(&rpcInit);
  if (pRpc == NULL) {
    uError("failed to start RPC server");
    return -1;
  }

  tsSyncPort = rpcInit.localPort + 10;
  qhandle = taosOpenQueue();

  doSync();

  pthread_attr_t thattr;
  pthread_t      thread;
  pthread_attr_init(&thattr);
  pthread_attr_setdetachstate(&thattr, PTHREAD_CREATE_JOINABLE);
  if (pthread_create(&thread, &thattr, processWriteQueue, NULL) != 0) {
    uError("failed to create thread, reason:%s", strerror(errno));
    return -1;
  }

  printf("server is running, localPort:%d\n", rpcInit.localPort);
  SNodesRole nroles;

  while (1) {
    char c = getchar();

    switch(c) {
      case '1':
        pCfg->replica = 1; doSync();
        break;        
      case '2':
        pCfg->replica = 2; doSync();
        break;
      case '3':
        pCfg->replica = 3; doSync();
        break;
      case '4':
        pCfg->replica = 4; doSync();
        break;
      case '5':
        pCfg->replica = 5; doSync();
        break;
      case 's':
        syncGetNodesRole(syncHandle, &nroles);
        for (int i=0; i<pCfg->replica; ++i) 
          printf("=== nodeId:%d role:%s\n", nroles.nodeId[i], syncRole[nroles.role[i]]);
        break;
      default:
        break;
    }

    if (c=='q') break;

  }

  syncStop(syncHandle);

  if (dataFd >= 0) {
    close(dataFd);
    remove(dataName);
  }

  return 0;
}