// highlight how much faster text movement between segments
// is when special casing Text and Data
//
// run this test with capn.go:1334-1341 commented in/out to compare.
//
func BenchmarkTextMovementBetweenSegments(b *testing.B) {

	buf := make([]byte, 1<<21)
	buf2 := make([]byte, 1<<21)

	text := make([]byte, 1<<20)
	for i := range text {
		text[i] = byte(65 + rand.Int()%26)
	}
	//stext := string(text)
	//fmt.Printf("text = %#v\n", stext)

	astr := make([]string, 1000)
	for i := range astr {
		astr[i] = string(text[i*1000 : (i+1)*1000])
	}

	b.ResetTimer()
	for i := 0; i < b.N; i++ {
		seg := capn.NewBuffer(buf[:0])
		scratch := capn.NewBuffer(buf2[:0])

		ht := air.NewRootHoldsText(seg)
		tl := scratch.NewTextList(1000)

		for j := 0; j < 1000; j++ {
			tl.Set(j, astr[j])
		}

		ht.SetLst(tl)

	}
}

func newTwoBTxnVotesSender(outcome *msgs.Outcome, txnId *common.TxnId, submitter common.RMId, recipients ...common.RMId) *twoBTxnVotesSender {
	submitterSeg := capn.NewBuffer(nil)
	submitterMsg := msgs.NewRootMessage(submitterSeg)
	submitterMsg.SetSubmissionOutcome(*outcome)

	if outcome.Which() == msgs.OUTCOME_ABORT {
		abort := outcome.Abort()
		abort.SetResubmit() // nuke out the updates as proposers don't need them.
	}

	seg := capn.NewBuffer(nil)
	msg := msgs.NewRootMessage(seg)
	twoB := msgs.NewTwoBTxnVotes(seg)
	msg.SetTwoBTxnVotes(twoB)
	twoB.SetOutcome(*outcome)

	server.Log(txnId, "Sending 2B to", recipients)

	return &twoBTxnVotesSender{
		msg:          server.SegToBytes(seg),
		recipients:   recipients,
		submitterMsg: server.SegToBytes(submitterSeg),
		submitter:    submitter,
	}
}

func TestObjectWithListOfTextBetweenSegments(t *testing.T) {

	exp := CapnpEncode(`(counter = (size = 9, wordlist = ["hello","bye"]))`, "Bag")

	cv.Convey("Given an Counter in one segment and a Bag with text in another", t, func() {
		cv.Convey("we should be able to copy from one segment to the other with SetCounter() on a Bag", func() {

			seg := capn.NewBuffer(nil)
			scratch := capn.NewBuffer(nil)

			// in seg
			segbag := air.NewRootBag(seg)

			// in scratch
			xc := air.NewRootCounter(scratch)
			xc.SetSize(9)
			tl := scratch.NewTextList(2)
			tl.Set(0, "hello")
			tl.Set(1, "bye")
			xc.SetWordlist(tl)

			xbuf := bytes.Buffer{}
			scratch.WriteTo(&xbuf)

			x := xbuf.Bytes()
			save(x, "myscratch")
			fmt.Printf("scratch segment (%p):\n", scratch)
			ShowBytes(x, 10)
			fmt.Printf("scratch segment (%p) with Counter decoded by capnp: '%s'\n", scratch, string(CapnpDecode(x, "Counter")))

			prebuf := bytes.Buffer{}
			seg.WriteTo(&prebuf)
			fmt.Printf("Bag only segment seg (%p), pre-transfer:\n", seg)
			ShowBytes(prebuf.Bytes(), 10)

			// now for the actual test:
			// copy from scratch to seg
			segbag.SetCounter(xc)

			buf := bytes.Buffer{}
			seg.WriteTo(&buf)

			act := buf.Bytes()
			save(act, "myact")
			save(exp, "myexp")

			fmt.Printf("expected:\n")
			ShowBytes(exp, 10)
			fmt.Printf("exp decoded by capnp: '%s'\n", string(CapnpDecode(exp, "Bag")))

			fmt.Printf("          actual:\n")
			ShowBytes(act, 10)
			fmt.Printf("act decoded by capnp: '%s'\n", string(CapnpDecode(act, "Bag")))

			cv.So(act, cv.ShouldResemble, exp)
		})
	})
}

func TestDataVersioningZeroPointersToMore(t *testing.T) {

	expEmpty := CapnpEncode("(mylist = [(),()])", "HoldsVerEmptyList")

	cv.Convey("Given a struct with 0 ptr fields, and a newer version of the struct with 1-2 pointer fields", t, func() {
		cv.Convey("then serializing the empty list and reading it back into 1 or 2 pointer fields should default initialize the pointer fields", func() {

			seg := capn.NewBuffer(nil)
			scratch := capn.NewBuffer(nil)

			emptyholder := air.NewRootHoldsVerEmptyList(seg)
			elist := air.NewVerEmptyList(scratch, 2)
			emptyholder.SetMylist(elist)

			actEmpty := ShowSeg("          after NewRootHoldsVerEmptyList(seg) and SetMylist(elist), segment seg is:", seg)
			actEmptyCap := string(CapnpDecode(actEmpty, "HoldsVerEmptyList"))
			expEmptyCap := string(CapnpDecode(expEmpty, "HoldsVerEmptyList"))
			cv.So(actEmptyCap, cv.ShouldResemble, expEmptyCap)

			fmt.Printf("\n actEmpty is \n")
			ShowBytes(actEmpty, 10)
			fmt.Printf("actEmpty decoded by capnp: '%s'\n", string(CapnpDecode(actEmpty, "HoldsVerEmptyList")))
			cv.So(actEmpty, cv.ShouldResemble, expEmpty)

			// seg is set, now read into bigger list
			buf := bytes.Buffer{}
			seg.WriteTo(&buf)
			segbytes := buf.Bytes()

			reseg, _, err := capn.ReadFromMemoryZeroCopy(segbytes)
			if err != nil {
				panic(err)
			}
			ShowSeg("      after re-reading segbytes, segment reseg is:", reseg)
			fmt.Printf("segbytes decoded by capnp as HoldsVerOneDataList: '%s'\n", string(CapnpDecode(segbytes, "HoldsVerOneDataList")))

			reHolder := air.ReadRootHoldsVerTwoTwoList(reseg)
			list22 := reHolder.Mylist()
			len22 := list22.Len()
			cv.So(len22, cv.ShouldEqual, 2)
			for i := 0; i < 2; i++ {
				ele := list22.At(i)
				val := ele.Val()
				cv.So(val, cv.ShouldEqual, 0)
				duo := ele.Duo()
				cv.So(duo, cv.ShouldEqual, 0)
				ptr1 := ele.Ptr1()
				ptr2 := ele.Ptr2()
				fmt.Printf("ptr1 = %#v\n", ptr1)
				cv.So(ptr1.Segment, cv.ShouldEqual, nil)
				fmt.Printf("ptr2 = %#v\n", ptr2)
				cv.So(ptr2.Segment, cv.ShouldEqual, nil)
			}

		})
	})
}

func TestZserverWithAccessors(t *testing.T) {

	exp := CapnpEncode(`(waitingjobs = [(cmd = "abc"), (cmd = "xyz")])`, "Zserver")

	cv.Convey("Given an Zserver with a custom list", t, func() {
		cv.Convey("then all the accessors should work as expected", func() {

			seg := capn.NewBuffer(nil)
			scratch := capn.NewBuffer(nil)

			server := air.NewRootZserver(seg)

			joblist := air.NewZjobList(seg, 2)

			// .Set(int, item)
			zjob := air.NewZjob(scratch)
			zjob.SetCmd("abc")
			joblist.Set(0, zjob)

			zjob = air.NewZjob(scratch)
			zjob.SetCmd("xyz")
			joblist.Set(1, zjob)

			// .At(int)
			cv.So(joblist.At(0).Cmd(), cv.ShouldEqual, "abc")
			cv.So(joblist.At(1).Cmd(), cv.ShouldEqual, "xyz")

			// .Len()
			cv.So(joblist.Len(), cv.ShouldEqual, 2)

			// .ToArray()
			cv.So(len(joblist.ToArray()), cv.ShouldEqual, 2)
			cv.So(joblist.ToArray()[0].Cmd(), cv.ShouldEqual, "abc")
			cv.So(joblist.ToArray()[1].Cmd(), cv.ShouldEqual, "xyz")

			server.SetWaitingjobs(joblist)

			buf := bytes.Buffer{}
			seg.WriteTo(&buf)

			act := buf.Bytes()
			fmt.Printf("          actual:\n")
			ShowBytes(act, 10)
			fmt.Printf("act decoded by capnp: '%s'\n", string(CapnpDecode(act, "Zserver")))
			save(act, "myact")

			fmt.Printf("expected:\n")
			ShowBytes(exp, 10)
			fmt.Printf("exp decoded by capnp: '%s'\n", string(CapnpDecode(exp, "Zserver")))
			save(exp, "myexp")

			cv.So(act, cv.ShouldResemble, exp)
		})
	})
}

func (awtd *acceptorWriteToDisk) start() {
	outcome := awtd.outcome
	outcomeCap := (*msgs.Outcome)(outcome)
	awtd.sendToAll = awtd.sendToAll || outcomeCap.Which() == msgs.OUTCOME_COMMIT
	sendToAll := awtd.sendToAll
	stateSeg := capn.NewBuffer(nil)
	state := msgs.NewRootAcceptorState(stateSeg)
	state.SetTxn(*awtd.ballotAccumulator.Txn)
	state.SetOutcome(*outcomeCap)
	state.SetSendToAll(awtd.sendToAll)
	state.SetInstances(awtd.ballotAccumulator.AddInstancesToSeg(stateSeg))

	data := server.SegToBytes(stateSeg)

	// to ensure correct order of writes, schedule the write from
	// the current go-routine...
	server.Log(awtd.txnId, "Writing 2B to disk...")
	future := awtd.acceptorManager.Disk.ReadWriteTransaction(false, func(rwtxn *mdbs.RWTxn) (interface{}, error) {
		return nil, rwtxn.Put(db.DB.BallotOutcomes, awtd.txnId[:], data, 0)
	})
	go func() {
		// ... but process the result in a new go-routine to avoid blocking the executor.
		if _, err := future.ResultError(); err != nil {
			log.Printf("Error: %v Acceptor Write error: %v", awtd.txnId, err)
			return
		}
		server.Log(awtd.txnId, "Writing 2B to disk...done.")
		awtd.acceptorManager.Exe.Enqueue(func() { awtd.writeDone(outcome, sendToAll) })
	}()
}

func zdateFilledSegment(n int, packed bool) (*capn.Segment, []byte) {
	seg := capn.NewBuffer(nil)
	z := air.NewRootZ(seg)
	list := air.NewZdateList(seg, n)
	// hand added a Set() method to messages_test.go, so plist not needed
	plist := capn.PointerList(list)

	for i := 0; i < n; i++ {
		d := air.NewZdate(seg)
		d.SetMonth(12)
		d.SetDay(7)
		d.SetYear(int16(2004 + i))
		plist.Set(i, capn.Object(d))
		//list.Set(i, d)
	}
	z.SetZdatevec(list)

	buf := bytes.Buffer{}
	if packed {
		seg.WriteToPacked(&buf)
	} else {
		seg.WriteTo(&buf)
	}
	return seg, buf.Bytes()
}

func (palc *proposerAwaitLocallyComplete) maybeWriteToDisk() {
	if !(palc.currentState == palc && palc.callbackInvoked && palc.allAcceptorsAgreed) {
		return
	}

	stateSeg := capn.NewBuffer(nil)
	state := msgs.NewRootProposerState(stateSeg)
	acceptorsCap := stateSeg.NewUInt32List(len(palc.acceptors))
	state.SetAcceptors(acceptorsCap)
	for idx, rmId := range palc.acceptors {
		acceptorsCap.Set(idx, uint32(rmId))
	}

	data := server.SegToBytes(stateSeg)

	future := palc.proposerManager.Disk.ReadWriteTransaction(false, func(rwtxn *mdbs.RWTxn) (interface{}, error) {
		return nil, rwtxn.Put(db.DB.Proposers, palc.txnId[:], data, 0)
	})
	go func() {
		if _, err := future.ResultError(); err != nil {
			log.Printf("Error: %v when writing proposer to disk: %v\n", palc.txnId, err)
			return
		}
		palc.proposerManager.Exe.Enqueue(palc.writeDone)
	}()
}

func sendZDate(nnzbus *nn.Socket) {

	// Create Zdate and Write to nnzbus.
	//
	// Z is our universal type, a discriminated union.
	// Hence Z provides run-time type identification
	// for all message types defined in zbus.capnp
	//
	s := capn.NewBuffer(nil)
	z := gozbus.NewRootZ(s)
	d := gozbus.NewRootZdate(s)
	z.SetZdate(d)

	d.SetYear(2004)
	d.SetMonth(12)
	d.SetDay(7)

	buf := bytes.Buffer{}
	s.WriteTo(&buf)

	nnzbus.Send(buf.Bytes(), 0)

	fmt.Printf("[pid %d] sendZDate sent ZDate message: year %d, month %d, day %d\n",
		os.Getpid(), d.Year(), d.Month(), d.Day())
}

func TestDataTextCopyOptimization(t *testing.T) {
	cv.Convey("Given a text list from a different segment", t, func() {
		cv.Convey("Adding it to a different segment shouldn't panic", func() {
			seg := capn.NewBuffer(nil)
			seg2 := capn.NewBuffer(nil)
			root := air.NewRootNester1Capn(seg)

			strsl := seg2.NewTextList(256)
			for i := 0; i < strsl.Len(); i++ {
				strsl.Set(i, "testess")
			}

			root.SetStrs(strsl)
		})
	})
}

func newQuasarDB(c *quasarConfig) *quasarDB {
	var err error
	q := &quasarDB{
		addr:           c.addr,
		mdStore:        c.mdStore,
		maxConnections: 200,
	}
	log.Noticef("Connecting to Quasar at %v...", q.addr.String())
	q.packetpool = sync.Pool{
		New: func() interface{} {
			seg := capn.NewBuffer(nil)
			req := qsr.NewRootRequest(seg)
			req.SetEchoTag(0)
			ins := qsr.NewCmdInsertValues(seg)
			ins.SetSync(false)
			return quasarReading{
				seg: seg,
				req: &req,
				ins: &ins,
			}
		},
	}
	q.bufferpool = sync.Pool{
		New: func() interface{} {
			return bytes.NewBuffer(make([]byte, 0, 200)) // 200 byte buffer
		},
	}
	if q.connpool, err = NewConnectionPool(q.getConnection, q.maxConnections); err != nil {
		log.Fatal(err)
	}
	return q
}

func (cr *connectionRun) start() (bool, error) {
	log.Printf("Connection established to %v (%v)\n", cr.remoteHost, cr.remoteRMId)

	seg := capn.NewBuffer(nil)
	message := msgs.NewRootMessage(seg)
	message.SetHeartbeat()
	cr.beatBytes = server.SegToBytes(seg)

	if cr.isServer {
		cr.connectionManager.ServerEstablished(cr.Connection)
	}
	if cr.isClient {
		topology, servers := cr.connectionManager.ClientEstablished(cr.ConnectionNumber, cr.Connection)
		cr.connectionManager.AddSender(cr.Connection)
		cr.submitter = client.NewClientTxnSubmitter(cr.connectionManager.RMId, cr.connectionManager.BootCount, topology, cr.connectionManager)
		cr.submitter.TopologyChange(nil, servers)
	}
	cr.mustSendBeat = true
	cr.missingBeats = 0

	cr.beater = newConnectionBeater(cr.Connection)
	go cr.beater.beat()

	cr.reader = newConnectionReader(cr.Connection)
	go cr.reader.read()

	return false, nil
}

func (quasar *quasarDB) queryNearestValue(uuids []common.UUID, start uint64, backwards bool) ([]common.SmapNumbersResponse, error) {
	var ret = make([]common.SmapNumbersResponse, len(uuids))
	conn := quasar.connpool.Get()
	defer quasar.connpool.Put(conn)
	for i, uu := range uuids {
		seg := capn.NewBuffer(nil)
		req := qsr.NewRootRequest(seg)
		qnv := qsr.NewCmdQueryNearestValue(seg)
		qnv.SetBackward(backwards)
		uuid, _ := uuid.FromString(string(uu))
		qnv.SetUuid(uuid.Bytes())
		qnv.SetTime(int64(start))
		req.SetQueryNearestValue(qnv)
		_, err := seg.WriteTo(conn) // here, ignoring # bytes written
		if err != nil {
			return ret, err
		}
		sr, err := quasar.receive(conn)
		if err != nil {
			return ret, err
		}
		sr.UUID = uu
		ret[i] = sr
	}
	return ret, nil
}

func (cah *connectionAwaitServerHandshake) makeHelloFromServer(topology *server.Topology) *capn.Segment {
	seg := capn.NewBuffer(nil)
	hello := msgs.NewRootHelloFromServer(seg)
	localHost := cah.connectionManager.LocalHost()
	hello.SetLocalHost(localHost)
	namespace := make([]byte, common.KeyLen-8)
	binary.BigEndian.PutUint32(namespace[0:4], cah.ConnectionNumber)
	binary.BigEndian.PutUint32(namespace[4:8], cah.connectionManager.BootCount)
	binary.BigEndian.PutUint32(namespace[8:], uint32(cah.connectionManager.RMId))
	hello.SetNamespace(namespace)
	if cah.isServer {
		tieBreak := cah.rng.Uint32()
		cah.Lock()
		cah.combinedTieBreak = tieBreak
		cah.Unlock()
		hello.SetTieBreak(tieBreak)
		hello.SetTopologyDBVersion(topology.DBVersion[:])
		hello.SetTopology(topology.AddToSegAutoRoot(seg))
	}
	if topology.RootVarUUId != nil {
		varIdPos := msgs.NewVarIdPos(seg)
		hello.SetRoot(varIdPos)
		varIdPos.SetId(topology.RootVarUUId[:])
		varIdPos.SetPositions((capn.UInt8List)(*topology.RootPositions))
	}
	return seg
}

func (cr *connectionRun) submitTxn(txnMsg *connectionMsgTxn) error {
	if cr.currentState != cr {
		if !txnMsg.setOutcomeError(nil, nil, fmt.Errorf("Connection in wrong state: %v", cr.currentState)) {
			return fmt.Errorf("Live txn already closed")
		}
		return nil
	}
	if cr.liveTxn != nil {
		if !txnMsg.setOutcomeError(nil, nil, fmt.Errorf("Existing live txn")) {
			return fmt.Errorf("Live txn already closed")
		}
		return nil
	}
	binary.BigEndian.PutUint64(cr.namespace[:8], cr.nextTxnId)
	txnMsg.txn.SetId(cr.namespace)
	seg := capn.NewBuffer(nil)
	msg := msgs.NewRootMessage(seg)
	msg.SetClientTxnSubmission(*txnMsg.txn)
	if err := cr.sendMessage(&msg); err == nil {
		cr.liveTxn = txnMsg
		return nil
	} else {
		cr.nextTxnId++
		return err
	}
}

func (cr *connectionRun) start() (bool, error) {
	log.Printf("Connection established to %v (%v)\n", cr.serverHost, cr.rmId)

	seg := capn.NewBuffer(nil)
	message := msgs.NewRootMessage(seg)
	message.SetHeartbeat()
	buf := new(bytes.Buffer)
	_, err := seg.WriteTo(buf)
	if err != nil {
		return false, err
	}
	cr.beatBytes = buf.Bytes()

	cr.mustSendBeat = true
	cr.missingBeats = 0

	cr.beater = newConnectionBeater(cr.Connection)
	go cr.beater.beat()

	cr.reader = newConnectionReader(cr.Connection)
	go cr.reader.read()

	if cr.awaiting != nil {
		close(cr.awaiting.resultChan)
		cr.awaiting = nil
	}

	return false, nil
}

func (am *AcceptorManager) OneATxnVotesReceived(sender common.RMId, txnId *common.TxnId, oneATxnVotes *msgs.OneATxnVotes) {
	instanceRMId := common.RMId(oneATxnVotes.RmId())
	server.Log(txnId, "1A received from", sender, "; instance:", instanceRMId)
	instId := instanceId([instanceIdLen]byte{})
	instIdSlice := instId[:]
	copy(instIdSlice, txnId[:])
	binary.BigEndian.PutUint32(instIdSlice[common.KeyLen:], uint32(instanceRMId))

	replySeg := capn.NewBuffer(nil)
	msg := msgs.NewRootMessage(replySeg)
	oneBTxnVotes := msgs.NewOneBTxnVotes(replySeg)
	msg.SetOneBTxnVotes(oneBTxnVotes)
	oneBTxnVotes.SetRmId(oneATxnVotes.RmId())
	oneBTxnVotes.SetTxnId(oneATxnVotes.TxnId())

	proposals := oneATxnVotes.Proposals()
	promises := msgs.NewTxnVotePromiseList(replySeg, proposals.Len())
	oneBTxnVotes.SetPromises(promises)
	for idx, l := 0, proposals.Len(); idx < l; idx++ {
		proposal := proposals.At(idx)
		vUUId := common.MakeVarUUId(proposal.VarId())
		copy(instIdSlice[common.KeyLen+4:], vUUId[:])
		promise := promises.At(idx)
		promise.SetVarId(vUUId[:])
		am.ensureInstance(txnId, &instId, vUUId).OneATxnVotesReceived(&proposal, &promise)
	}

	NewOneShotSender(server.SegToBytes(replySeg), am.ConnectionManager, sender)
}

func (txn *Txn) submitRetryTransaction() error {
	reads := make(map[common.VarUUId]*objectState)
	for ancestor := txn; ancestor != nil; ancestor = ancestor.parent {
		for _, obj := range ancestor.objs {
			if _, found := reads[*obj.Id]; !found && obj.state.txn == ancestor && obj.state.read {
				reads[*obj.Id] = obj.state
			}
		}
	}
	seg := capn.NewBuffer(nil)
	cTxn := msgs.NewClientTxn(seg)
	cTxn.SetRetry(true)
	actions := msgs.NewClientActionList(seg, len(reads))
	cTxn.SetActions(actions)
	idx := 0
	for _, state := range reads {
		action := actions.At(idx)
		action.SetVarId(state.Id[:])
		action.SetRead()
		action.Read().SetVersion(state.curVersion[:])
		idx++
	}
	outcome, _, err := txn.conn.submitTransaction(&cTxn)
	if err != nil {
		return err
	}
	txn.stats.TxnId = common.MakeTxnId(outcome.FinalId())
	for ancestor := txn; ancestor != nil; ancestor = ancestor.parent {
		ancestor.resetInProgress = true
	}
	return nil
}

func (cr *connectionRun) handleMsgFromPeer(msg *msgs.Message) error {
	if cr.currentState != cr {
		// probably just draining the queue from the reader after a restart
		return nil
	}
	cr.missingBeats = 0
	switch which := msg.Which(); which {
	case msgs.MESSAGE_HEARTBEAT:
		// do nothing
	case msgs.MESSAGE_CLIENTTXNSUBMISSION:
		ctxn := msg.ClientTxnSubmission()
		origTxnId := common.MakeTxnId(ctxn.Id())
		cr.submitter.SubmitClientTransaction(&ctxn, func(clientOutcome *msgs.ClientTxnOutcome, err error) {
			switch {
			case err != nil:
				cr.clientTxnError(&ctxn, err, origTxnId)
			case clientOutcome == nil: // shutdown
				return
			default:
				seg := capn.NewBuffer(nil)
				msg := msgs.NewRootMessage(seg)
				msg.SetClientTxnOutcome(*clientOutcome)
				cr.sendMessage(server.SegToBytes(msg.Segment))
			}
		})
	default:
		cr.connectionManager.Dispatchers.DispatchMessage(cr.remoteRMId, which, msg)
	}
	return nil
}

//TODO: fix?
func (q *quasarDB) WindowData(uuids []common.UUID, pointWidth, start, end uint64) ([]common.StatisticalNumbersResponse, error) {
	var ret = make([]common.StatisticalNumbersResponse, len(uuids))
	conn := q.connpool.Get()
	defer q.connpool.Put(conn)
	for i, uu := range uuids {
		seg := capn.NewBuffer(nil)
		req := qsr.NewRootRequest(seg)
		query := qsr.NewCmdQueryStatisticalValues(seg)
		uuid, _ := uuid.FromString(string(uu))
		query.SetUuid(uuid.Bytes())
		query.SetStartTime(int64(start))
		query.SetEndTime(int64(end))
		query.SetPointWidth(uint8(pointWidth))
		req.SetQueryStatisticalValues(query)
		_, err := seg.WriteTo(conn) // here, ignoring # bytes written
		if err != nil {
			return ret, err
		}
		sr, err := q.receiveStats(conn)
		if err != nil {
			return ret, err
		}
		sr.UUID = uu
		ret[i] = sr
	}
	return ret, nil
}