// pushTxn attempts to abort the txn via push. If the transaction
// cannot be aborted, the oldestIntentNanos value is atomically
// updated to the min of oldestIntentNanos and the intent's
// timestamp. The wait group is signaled on completion.
func (gcq *gcQueue) pushTxn(repl *Replica, now proto.Timestamp, txn *proto.Transaction, updateOldestIntent func(int64), wg *sync.WaitGroup) {
	defer wg.Done() // signal wait group always on completion
	if log.V(1) {
		log.Infof("pushing txn %s ts=%s", txn, txn.OrigTimestamp)
	}

	// Attempt to push the transaction which created the intent.
	ba := &proto.BatchRequest{}
	ba.Timestamp = now
	ba.UserPriority = gogoproto.Int32(proto.MaxPriority)
	pushArgs := &proto.PushTxnRequest{
		RequestHeader: proto.RequestHeader{
			Timestamp:    now,
			Key:          txn.Key,
			UserPriority: gogoproto.Int32(proto.MaxPriority),
		},
		Now:       now,
		PusherTxn: nil,
		PusheeTxn: *txn,
		PushType:  proto.ABORT_TXN,
	}
	ba.Add(pushArgs)
	b := &client.Batch{}
	b.InternalAddRequest(ba)
	br, err := repl.rm.DB().RunWithResponse(b)
	if err != nil {
		log.Warningf("push of txn %s failed: %s", txn, err)
		updateOldestIntent(txn.OrigTimestamp.WallTime)
		return
	}
	// Update the supplied txn on successful push.
	*txn = *br.Responses[0].GetInner().(*proto.PushTxnResponse).PusheeTxn
}

func TestUnmarshalPartiallyPopulatedOptionalFieldsFails(t *testing.T) {
	// Fill in all fields, then randomly remove one.
	dataOut := &test.NinOptNative{
		Field1:  proto.Float64(0),
		Field2:  proto.Float32(0),
		Field3:  proto.Int32(0),
		Field4:  proto.Int64(0),
		Field5:  proto.Uint32(0),
		Field6:  proto.Uint64(0),
		Field7:  proto.Int32(0),
		Field8:  proto.Int64(0),
		Field9:  proto.Uint32(0),
		Field10: proto.Int32(0),
		Field11: proto.Uint64(0),
		Field12: proto.Int64(0),
		Field13: proto.Bool(false),
		Field14: proto.String("0"),
		Field15: []byte("0"),
	}
	r := rand.New(rand.NewSource(time.Now().UnixNano()))
	fieldName := "Field" + strconv.Itoa(r.Intn(15)+1)
	field := reflect.ValueOf(dataOut).Elem().FieldByName(fieldName)
	fieldType := field.Type()
	field.Set(reflect.Zero(fieldType))
	encodedMessage, err := proto.Marshal(dataOut)
	if err != nil {
		t.Fatalf("Unexpected error when marshalling dataOut: %v", err)
	}
	dataIn := NidOptNative{}
	err = proto.Unmarshal(encodedMessage, &dataIn)
	if err.Error() != `proto: required field "`+fieldName+`" not set` {
		t.Fatalf(`err.Error() != "proto: required field "`+fieldName+`" not set"; was "%s" instead`, err.Error())
	}
}

func createStartStopMessage(requestId uint64, peerType events.PeerType) *events.Envelope {
	return &events.Envelope{
		Origin:    proto.String("fake-origin-2"),
		EventType: events.Envelope_HttpStartStop.Enum(),
		HttpStartStop: &events.HttpStartStop{
			StartTimestamp: proto.Int64(1),
			StopTimestamp:  proto.Int64(100),
			RequestId: &events.UUID{
				Low:  proto.Uint64(requestId),
				High: proto.Uint64(requestId + 1),
			},
			PeerType:      &peerType,
			Method:        events.Method_GET.Enum(),
			Uri:           proto.String("fake-uri-1"),
			RemoteAddress: proto.String("fake-remote-addr-1"),
			UserAgent:     proto.String("fake-user-agent-1"),
			StatusCode:    proto.Int32(103),
			ContentLength: proto.Int64(104),
			ParentRequestId: &events.UUID{
				Low:  proto.Uint64(2),
				High: proto.Uint64(3),
			},
			ApplicationId: &events.UUID{
				Low:  proto.Uint64(105),
				High: proto.Uint64(106),
			},
			InstanceIndex: proto.Int32(6),
			InstanceId:    proto.String("fake-instance-id-1"),
		},
	}
}

func newTestMessage() *pb.MyMessage {
	msg := &pb.MyMessage{
		Count: proto.Int32(42),
		Name:  proto.String("Dave"),
		Quote: proto.String(`"I didn't want to go."`),
		Pet:   []string{"bunny", "kitty", "horsey"},
		Inner: &pb.InnerMessage{
			Host:      proto.String("footrest.syd"),
			Port:      proto.Int32(7001),
			Connected: proto.Bool(true),
		},
		Others: []*pb.OtherMessage{
			{
				Key:   proto.Int64(0xdeadbeef),
				Value: []byte{1, 65, 7, 12},
			},
			{
				Weight: proto.Float32(6.022),
				Inner: &pb.InnerMessage{
					Host: proto.String("lesha.mtv"),
					Port: proto.Int32(8002),
				},
			},
		},
		Bikeshed: pb.MyMessage_BLUE.Enum(),
		Somegroup: &pb.MyMessage_SomeGroup{
			GroupField: proto.Int32(8),
		},
		// One normally wouldn't do this.
		// This is an undeclared tag 13, as a varint (wire type 0) with value 4.
		XXX_unrecognized: []byte{13<<3 | 0, 4},
	}
	ext := &pb.Ext{
		Data: proto.String("Big gobs for big rats"),
	}
	if err := proto.SetExtension(msg, pb.E_Ext_More, ext); err != nil {
		panic(err)
	}
	greetings := []string{"adg", "easy", "cow"}
	if err := proto.SetExtension(msg, pb.E_Greeting, greetings); err != nil {
		panic(err)
	}

	// Add an unknown extension. We marshal a pb.Ext, and fake the ID.
	b, err := proto.Marshal(&pb.Ext{Data: proto.String("3G skiing")})
	if err != nil {
		panic(err)
	}
	b = append(proto.EncodeVarint(201<<3|proto.WireBytes), b...)
	proto.SetRawExtension(msg, 201, b)

	// Extensions can be plain fields, too, so let's test that.
	b = append(proto.EncodeVarint(202<<3|proto.WireVarint), 19)
	proto.SetRawExtension(msg, 202, b)

	return msg
}

func encodeAux(aux []interface{}) []*internal.Aux {
	pb := make([]*internal.Aux, len(aux))
	for i := range aux {
		switch v := aux[i].(type) {
		case float64:
			pb[i] = &internal.Aux{DataType: proto.Int32(Float), FloatValue: proto.Float64(v)}
		case *float64:
			pb[i] = &internal.Aux{DataType: proto.Int32(Float)}
		case int64:
			pb[i] = &internal.Aux{DataType: proto.Int32(Integer), IntegerValue: proto.Int64(v)}
		case *int64:
			pb[i] = &internal.Aux{DataType: proto.Int32(Integer)}
		case string:
			pb[i] = &internal.Aux{DataType: proto.Int32(String), StringValue: proto.String(v)}
		case *string:
			pb[i] = &internal.Aux{DataType: proto.Int32(String)}
		case bool:
			pb[i] = &internal.Aux{DataType: proto.Int32(Boolean), BooleanValue: proto.Bool(v)}
		case *bool:
			pb[i] = &internal.Aux{DataType: proto.Int32(Boolean)}
		default:
			pb[i] = &internal.Aux{DataType: proto.Int32(int32(Unknown))}
		}
	}
	return pb
}

func ExampleCompile() {
	a := &test.NinOptNative{
		Field4: proto.Int64(1234),
		Field7: proto.Int32(123),
	}
	fp1, err := fieldpath.NewInt64Path("test", "NinOptNative", test.ThetestDescription(), "Field4")
	if err != nil {
		panic(err)
	}
	fp2, err := fieldpath.NewSint32Path("test", "NinOptNative", test.ThetestDescription(), "Field7")
	if err != nil {
		panic(err)
	}
	buf, err := proto.Marshal(a)
	if err != nil {
		panic(err)
	}
	u1 := fieldpath.NewInt64Unmarshaler(fp1, &handler64{})
	u2 := fieldpath.NewSint32Unmarshaler(fp2, &handler32{})
	c := fieldpath.Compile(u1, u2)
	err = c.Unmarshal(buf)
	if err != nil {
		panic(err)
	}
	// Output:
	// 1234
	// 123
}

func (rp *ResponsePlotter) maybeConsolidateData(numberOfPixels float64) {
	// idealy numberOfPixels should be size in pixels of char ares,
	// not char areay with Y axis and its label

	numberOfDataPoints := len(rp.Response.Values)
	pointsPerPixel := int(math.Ceil(float64(numberOfDataPoints) / numberOfPixels))

	if pointsPerPixel <= 1 {
		return
	}

	newNumberOfDataPoints := (numberOfDataPoints / pointsPerPixel) + 1
	values := make([]float64, newNumberOfDataPoints)
	absent := make([]bool, newNumberOfDataPoints)

	k := 0
	step := pointsPerPixel
	for i := 0; i < numberOfDataPoints; i += step {
		if i+step < numberOfDataPoints {
			values[k], absent[k] = consolidateAvg(rp.Response.Values[i:i+step], rp.Response.IsAbsent[i:i+step])
		} else {
			values[k], absent[k] = consolidateAvg(rp.Response.Values[i:], rp.Response.IsAbsent[i:])
		}

		k++
	}

	stepTime := rp.Response.GetStepTime()
	stepTime *= int32(pointsPerPixel)

	rp.Response.Values = values[:k]
	rp.Response.IsAbsent = absent[:k]
	rp.Response.StepTime = proto.Int32(stepTime)
}

// TestTxnCoordSenderBeginTransactionMinPriority verifies that when starting
// a new transaction, a non-zero priority is treated as a minimum value.
func TestTxnCoordSenderBeginTransactionMinPriority(t *testing.T) {
	defer leaktest.AfterTest(t)
	s := createTestDB(t)
	defer s.Stop()
	defer teardownHeartbeats(s.Sender)

	reply := &proto.PutResponse{}
	s.Sender.Send(context.Background(), proto.Call{
		Args: &proto.PutRequest{
			RequestHeader: proto.RequestHeader{
				Key:          proto.Key("key"),
				User:         security.RootUser,
				UserPriority: gogoproto.Int32(-10), // negative user priority is translated into positive priority
				Txn: &proto.Transaction{
					Name:      "test txn",
					Isolation: proto.SNAPSHOT,
					Priority:  11,
				},
			},
		},
		Reply: reply,
	})
	if reply.Error != nil {
		t.Fatal(reply.GoError())
	}
	if reply.Txn.Priority != 11 {
		t.Errorf("expected txn priority 11; got %d", reply.Txn.Priority)
	}
}

func TestGetExtensionStability(t *testing.T) {
	check := func(m *pb.MyMessage) bool {
		ext1, err := proto.GetExtension(m, pb.E_Ext_More)
		if err != nil {
			t.Fatalf("GetExtension() failed: %s", err)
		}
		ext2, err := proto.GetExtension(m, pb.E_Ext_More)
		if err != nil {
			t.Fatalf("GetExtension() failed: %s", err)
		}
		return ext1 == ext2
	}
	msg := &pb.MyMessage{Count: proto.Int32(4)}
	ext0 := &pb.Ext{}
	if err := proto.SetExtension(msg, pb.E_Ext_More, ext0); err != nil {
		t.Fatalf("Could not set ext1: %s", ext0)
	}
	if !check(msg) {
		t.Errorf("GetExtension() not stable before marshaling")
	}
	bb, err := proto.Marshal(msg)
	if err != nil {
		t.Fatalf("Marshal() failed: %s", err)
	}
	msg1 := &pb.MyMessage{}
	err = proto.Unmarshal(bb, msg1)
	if err != nil {
		t.Fatalf("Unmarshal() failed: %s", err)
	}
	if !check(msg1) {
		t.Errorf("GetExtension() not stable after unmarshaling")
	}
}

func (GpbrpcType) OK(args ...interface{}) gpbrpc.ResultT {
	if len(args) == 0 {
		return okResult
	}
	return gpbrpc.Result(&ResponseGeneric{ErrorCode: proto.Int32(0),
		ErrorText: proto.String(fmt.Sprint(args...))})
}

// send runs the specified calls synchronously in a single batch and
// returns any errors.
func (db *DB) send(reqs ...proto.Request) (*proto.BatchResponse, *proto.Error) {
	if len(reqs) == 0 {
		return &proto.BatchResponse{}, nil
	}

	if len(reqs) == 1 {
		// We only send BatchRequest. Everything else needs to go into one.
		if ba, ok := reqs[0].(*proto.BatchRequest); ok {
			if ba.UserPriority == nil && db.userPriority != 0 {
				ba.UserPriority = gogoproto.Int32(db.userPriority)
			}
			resetClientCmdID(ba)
			br, pErr := db.sender.Send(context.TODO(), *ba)
			if pErr != nil {
				if log.V(1) {
					log.Infof("failed %s: %s", ba.Method(), pErr)
				}
				return nil, pErr
			}
			return br, nil
		}
	}

	ba := proto.BatchRequest{}
	ba.Add(reqs...)

	br, pErr := db.send(&ba)

	if pErr != nil {
		return nil, pErr
	}
	return br, nil
}

// TestTxnCoordSenderBeginTransactionMinPriority verifies that when starting
// a new transaction, a non-zero priority is treated as a minimum value.
func TestTxnCoordSenderBeginTransactionMinPriority(t *testing.T) {
	defer leaktest.AfterTest(t)
	s := createTestDB(t)
	defer s.Stop()
	defer teardownHeartbeats(s.Sender)

	reply, err := client.SendWrappedWith(s.Sender, nil, roachpb.BatchRequest_Header{
		UserPriority: proto.Int32(-10), // negative user priority is translated into positive priority
		Txn: &roachpb.Transaction{
			Name:      "test txn",
			Isolation: roachpb.SNAPSHOT,
			Priority:  11,
		},
	}, &roachpb.PutRequest{
		RequestHeader: roachpb.RequestHeader{
			Key: roachpb.Key("key"),
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	if prio := reply.(*roachpb.PutResponse).Txn.Priority; prio != 11 {
		t.Errorf("expected txn priority 11; got %d", prio)
	}
}

func newTestSender(handler func(proto.Call)) SenderFunc {
	txnKey := proto.Key("test-txn")
	txnID := []byte(uuid.NewUUID4())

	return func(_ context.Context, call proto.Call) {
		header := call.Args.Header()
		header.UserPriority = gogoproto.Int32(-1)
		if header.Txn != nil && len(header.Txn.ID) == 0 {
			header.Txn.Key = txnKey
			header.Txn.ID = txnID
		}
		call.Reply.Reset()
		var writing bool
		switch call.Args.(type) {
		case *proto.PutRequest:
			gogoproto.Merge(call.Reply, testPutResp)
			writing = true
		case *proto.EndTransactionRequest:
			writing = true
		default:
			// Do nothing.
		}
		call.Reply.Header().Txn = gogoproto.Clone(call.Args.Header().Txn).(*proto.Transaction)
		if txn := call.Reply.Header().Txn; txn != nil {
			txn.Writing = writing
		}

		if handler != nil {
			handler(call)
		}
	}
}

func TestMarshalRace(t *testing.T) {
	// unregistered extension
	desc := &proto.ExtensionDesc{
		ExtendedType:  (*pb.MyMessage)(nil),
		ExtensionType: (*bool)(nil),
		Field:         101010100,
		Name:          "emptyextension",
		Tag:           "varint,0,opt",
	}

	m := &pb.MyMessage{Count: proto.Int32(4)}
	if err := proto.SetExtension(m, desc, proto.Bool(true)); err != nil {
		t.Errorf("proto.SetExtension(m, desc, true): got error %q, want nil", err)
	}

	errChan := make(chan error, 3)
	for n := 3; n > 0; n-- {
		go func() {
			_, err := proto.Marshal(m)
			errChan <- err
		}()
	}
	for i := 0; i < 3; i++ {
		err := <-errChan
		if err != nil {
			t.Fatal(err)
		}
	}
}

// TestTxnCoordSenderBeginTransaction verifies that a command sent with a
// not-nil Txn with empty ID gets a new transaction initialized.
func TestTxnCoordSenderBeginTransaction(t *testing.T) {
	defer leaktest.AfterTest(t)
	s := createTestDB(t)
	defer s.Stop()
	defer teardownHeartbeats(s.Sender)

	key := proto.Key("key")
	reply, err := batchutil.SendWrapped(s.Sender, &proto.PutRequest{
		RequestHeader: proto.RequestHeader{
			Key:          key,
			UserPriority: gogoproto.Int32(-10), // negative user priority is translated into positive priority
			Txn: &proto.Transaction{
				Name:      "test txn",
				Isolation: proto.SNAPSHOT,
			},
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	pr := reply.(*proto.PutResponse)
	if pr.Txn.Name != "test txn" {
		t.Errorf("expected txn name to be %q; got %q", "test txn", pr.Txn.Name)
	}
	if pr.Txn.Priority != 10 {
		t.Errorf("expected txn priority 10; got %d", pr.Txn.Priority)
	}
	if !bytes.Equal(pr.Txn.Key, key) {
		t.Errorf("expected txn Key to match %q != %q", key, pr.Txn.Key)
	}
	if pr.Txn.Isolation != proto.SNAPSHOT {
		t.Errorf("expected txn isolation to be SNAPSHOT; got %s", pr.Txn.Isolation)
	}
}

func (d *LauncherData) terminate(row *RunQueueEntry, action string) {
	if d.killedRecently[row.Id] {
		return
	}

	if action == KILL_ACTION_NO_ACTION {
		d.call(&badoo_phproxyd.RequestTerminate{Hash: proto.Uint64(row.Id)})
	} else {
		params := []string{
			`\ScriptFramework\Script_Kill`,
			fmt.Sprintf("--force-sf-db=%s", db.GetDbName()),
			fmt.Sprintf("--kill-run-id=%d", row.Id),
			fmt.Sprintf("--kill-action=%s", action),
			fmt.Sprintf("--kill-class-name=%s", row.ClassName),
			fmt.Sprintf("--kill-timetable-id=%d", row.timetable_id.Int64),
		}

		d.call(&badoo_phproxyd.RequestRun{
			Script:       proto.String(getScriptPath(row.settings)),
			Hash:         proto.Uint64(0),
			Tag:          proto.String(PHPROXY_TAG),
			Force:        proto.Int32(1),
			Params:       params,
			Store:        badoo_phproxyd.StoreT_MEMORY.Enum(),
			FreeAfterRun: proto.Bool(true),
		})
	}

	d.killedRecently[row.Id] = true
}

// pushTxn attempts to abort the txn via push. If the transaction
// cannot be aborted, the oldestIntentNanos value is atomically
// updated to the min of oldestIntentNanos and the intent's
// timestamp. The wait group is signaled on completion.
func (gcq *gcQueue) pushTxn(repl *Replica, now proto.Timestamp, txn *proto.Transaction, updateOldestIntent func(int64), wg *sync.WaitGroup) {
	defer wg.Done() // signal wait group always on completion
	if log.V(1) {
		log.Infof("pushing txn %s ts=%s", txn, txn.OrigTimestamp)
	}

	// Attempt to push the transaction which created the intent.
	pushArgs := &proto.PushTxnRequest{
		RequestHeader: proto.RequestHeader{
			Timestamp:    now,
			Key:          txn.Key,
			User:         security.RootUser,
			UserPriority: gogoproto.Int32(proto.MaxPriority),
			Txn:          nil,
		},
		Now:       now,
		PusheeTxn: *txn,
		PushType:  proto.ABORT_TXN,
	}
	pushReply := &proto.PushTxnResponse{}
	b := &client.Batch{}
	b.InternalAddCall(proto.Call{Args: pushArgs, Reply: pushReply})
	if err := repl.rm.DB().Run(b); err != nil {
		log.Warningf("push of txn %s failed: %s", txn, err)
		updateOldestIntent(txn.OrigTimestamp.WallTime)
		return
	}
	// Update the supplied txn on successful push.
	*txn = *pushReply.PusheeTxn
}

func NewError(source string, code int32, message string) *events.Error {
	err := &events.Error{
		Source:  proto.String(source),
		Code:    proto.Int32(code),
		Message: proto.String(message),
	}
	return err
}

func (c *Client) SetPrivilege(username, database string, p influxql.Privilege) error {
	return c.retryUntilExec(internal.Command_SetPrivilegeCommand, internal.E_SetPrivilegeCommand_Command,
		&internal.SetPrivilegeCommand{
			Username:  proto.String(username),
			Database:  proto.String(database),
			Privilege: proto.Int32(int32(p)),
		},
	)
}

// SetPrivilege sets a privilege for a user on a database.
func (s *Store) SetPrivilege(username, database string, p influxql.Privilege) error {
	return s.exec(internal.Command_SetPrivilegeCommand, internal.E_SetPrivilegeCommand_Command,
		&internal.SetPrivilegeCommand{
			Username:  proto.String(username),
			Database:  proto.String(database),
			Privilege: proto.Int32(int32(p)),
		},
	)
}