text: /* -*- Mode: C++; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */
  /*
   *     Copyright 2011-2020 Couchbase, Inc.
   *
   *   Licensed under the Apache License, Version 2.0 (the "License");
   *   you may not use this file except in compliance with the License.
   *   You may obtain a copy of the License at
   *
   *       http://www.apache.org/licenses/LICENSE-2.0
   *
   *   Unless required by applicable law or agreed to in writing, software
   *   distributed under the License is distributed on an "AS IS" BASIS,
   *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   *   See the License for the specific language governing permissions and
   *   limitations under the License.
   */
  
  #include "iotests.h"
  #include "internalstructs.h"
  
  using namespace std;
  class ObseqnoTest : public MockUnitTest
  {
  };
  
  extern "C" {
  static void storeCb_getstok(lcb_INSTANCE *, int cbtype, const lcb_RESPSTORE *resp)
  {
      EXPECT_EQ(LCB_SUCCESS, lcb_respstore_status(resp));
      lcb_MUTATION_TOKEN *res = NULL;
      lcb_respstore_cookie(resp, (void **)&res);
      lcb_respstore_mutation_token(resp, res);
  }
  }
  
  static void storeGetStok(lcb_INSTANCE *instance, const string &k, const string &v, lcb_MUTATION_TOKEN *res)
  {
      lcb_RESPCALLBACK oldcb = lcb_get_callback(instance, LCB_CALLBACK_STORE);
      lcb_install_callback(instance, LCB_CALLBACK_STORE, (lcb_RESPCALLBACK)storeCb_getstok);
      lcb_sched_enter(instance);
  
      lcb_CMDSTORE *cmd;
      lcb_cmdstore_create(&cmd, LCB_STORE_UPSERT);
      lcb_cmdstore_key(cmd, k.c_str(), k.size());
      lcb_cmdstore_value(cmd, v.c_str(), v.size());
  
      lcb_STATUS rc = lcb_store(instance, res, cmd);
      EXPECT_EQ(LCB_SUCCESS, rc);
      lcb_cmdstore_destroy(cmd);
      lcb_sched_leave(instance);
      lcb_wait(instance, LCB_WAIT_DEFAULT);
      lcb_install_callback(instance, LCB_CALLBACK_STORE, oldcb);
  }
  
  TEST_F(ObseqnoTest, testFetchImplicit)
  {
      SKIP_UNLESS_MOCK();
      HandleWrap hw;
      lcb_INSTANCE *instance;
      lcb_STATUS rc;
      createConnection(hw, &instance);
      const char *key = "obseqBasic";
      const char *value = "value";
  
      lcb_MUTATION_TOKEN st_fetched = {0};
      storeGetStok(instance, key, value, &st_fetched);
      ASSERT_TRUE(st_fetched.uuid_ != 0);
  
      lcb_KEYBUF kb;
      LCB_KREQ_SIMPLE(&kb, key, strlen(key));
  
      const lcb_MUTATION_TOKEN *ss = lcb_get_mutation_token(instance, &kb, &rc);
      ASSERT_EQ(LCB_SUCCESS, rc);
      ASSERT_TRUE(ss != NULL);
      st_fetched = *ss;
  }
  
  extern "C" {
  static void obseqCallback(lcb_INSTANCE *, int, const lcb_RESPOBSEQNO *rb)
  {
      lcb_RESPOBSEQNO *pp = (lcb_RESPOBSEQNO *)rb->cookie;
      *pp = *rb;
  }
  }
  
  static void doObserveSeqno(lcb_INSTANCE *instance, lcb_MUTATION_TOKEN *ss, int server, lcb_RESPOBSEQNO &resp)
  {
      lcb_CMDOBSEQNO cmd = {0};
      cmd.vbid = ss->vbid_;
      cmd.uuid = ss->uuid_;
      cmd.server_index = server;
      lcb_STATUS rc;
  
      lcb_sched_enter(instance);
      rc = lcb_observe_seqno3(instance, &resp, &cmd);
      if (rc != LCB_SUCCESS) {
          resp.ctx.rc = rc;
          resp.rflags |= LCB_RESP_F_CLIENTGEN;
          return;
      }
  
      lcb_RESPCALLBACK oldcb = lcb_get_callback(instance, LCB_CALLBACK_OBSEQNO);
      lcb_install_callback(instance, LCB_CALLBACK_OBSEQNO, (lcb_RESPCALLBACK)obseqCallback);
      lcb_sched_leave(instance);
      lcb_wait(instance, LCB_WAIT_DEFAULT);
      lcb_install_callback(instance, LCB_CALLBACK_OBSEQNO, oldcb);
  }
  
  TEST_F(ObseqnoTest, testObserve)
  {
      SKIP_UNLESS_MOCK();
  
      HandleWrap hw;
      lcb_INSTANCE *instance;
      createConnection(hw, &instance);
      lcbvb_CONFIG *vbc;
  
      lcb_cntl(instance, LCB_CNTL_GET, LCB_CNTL_VBCONFIG, &vbc);
  
      lcb_MUTATION_TOKEN st_fetched = {0};
      const char *key = "testObserve";
      const char *value = "value";
  
      // Get the synctoken
      storeGetStok(instance, key, value, &st_fetched);
      ASSERT_NE(0, st_fetched.vbid_);
      ASSERT_NE(0, st_fetched.uuid_);
      ASSERT_NE(0, st_fetched.seqno_);
  
      for (size_t ii = 0; ii < lcbvb_get_nreplicas(vbc) + 1; ii++) {
          int ix = lcbvb_vbserver(vbc, st_fetched.vbid_, ii);
          lcb_RESPOBSEQNO resp{};
          doObserveSeqno(instance, &st_fetched, ix, resp);
          ASSERT_EQ(LCB_SUCCESS, resp.ctx.rc);
          ASSERT_EQ(st_fetched.uuid_, resp.cur_uuid);
          ASSERT_EQ(0, resp.old_uuid);
          //        printf("SEQ_MEM: %lu. SEQ_DISK: %lu\n", resp.mem_seqno, resp.persisted_seqno);
          ASSERT_GT(resp.mem_seqno, 0);
          ASSERT_EQ(resp.mem_seqno, resp.persisted_seqno);
      }
  }
  
  TEST_F(ObseqnoTest, testFailoverFormat)
  {
      SKIP_UNLESS_MOCK();
      HandleWrap hw;
      lcb_INSTANCE *instance;
      createConnection(hw, &instance);
      lcbvb_CONFIG *vbc;
  
      lcb_cntl(instance, LCB_CNTL_GET, LCB_CNTL_VBCONFIG, &vbc);
      const char *key = "testObserve";
      const char *value = "value";
  
      lcb_MUTATION_TOKEN st_fetched = {0};
      storeGetStok(instance, key, value, &st_fetched);
  
      MockEnvironment *env = MockEnvironment::getInstance();
      env->regenVbCoords();
  
      // Now we should get a different sequence number
      lcb_RESPOBSEQNO rr;
      doObserveSeqno(instance, &st_fetched, lcbvb_vbmaster(vbc, st_fetched.vbid_), rr);
      ASSERT_EQ(LCB_SUCCESS, rr.ctx.rc);
      //    printf("Old UUID: %llu\n", rr.old_uuid);
      //    printf("Cur UUID: %llu\n", rr.cur_uuid);
      ASSERT_GT(rr.old_uuid, 0);
      ASSERT_EQ(rr.old_uuid, st_fetched.uuid_);
      ASSERT_NE(rr.old_uuid, rr.cur_uuid);
      ASSERT_EQ(rr.old_seqno, st_fetched.seqno_);
  }
  
  // TODO: We should add more tests here, but in order to do this, we must
  // first validate the mock.
  