// Begin begins a transaction, and returns the associated transaction id.
// Subsequent statements can access the connection through the transaction id.
func (axp *TxPool) Begin(ctx context.Context) int64 {
	poolCtx := ctx
	if deadline, ok := ctx.Deadline(); ok {
		var cancel func()
		poolCtx, cancel = context.WithDeadline(ctx, deadline.Add(-10*time.Millisecond))
		defer cancel()
	}
	conn, err := axp.pool.Get(poolCtx)
	if err != nil {
		switch err {
		case ErrConnPoolClosed:
			panic(err)
		case pools.ErrTimeout:
			axp.LogActive()
			panic(NewTabletError(ErrTxPoolFull, "Transaction pool connection limit exceeded"))
		}
		panic(NewTabletErrorSql(ErrFatal, err))
	}
	if _, err := conn.Exec(ctx, "begin", 1, false); err != nil {
		conn.Recycle()
		panic(NewTabletErrorSql(ErrFail, err))
	}
	transactionID := axp.lastID.Add(1)
	axp.activePool.Register(transactionID, newTxConnection(conn, transactionID, axp))
	return transactionID
}

func TestContextCancel(t *testing.T) {
	// server that doesn't reply before the timeout
	wg := sync.WaitGroup{}
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		time.Sleep(2 * time.Second)
		fmt.Fprint(w, r.URL.Path)
		wg.Done()
	}))
	defer srv.Close()

	tr := NewTransport(nil, RetryAll(RetryMaxRetries(1), RetryStatusInterval(500, 600)), ConstDelay(0))
	c := &http.Client{
		Transport: tr,
	}

	ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(time.Second))
	defer cancelFn()

	wg.Add(1)
	res, err := ctxhttp.Get(ctx, c, srv.URL+"/test")
	require.Nil(t, res)

	assert.Equal(t, context.DeadlineExceeded, err)
	wg.Wait()
}

// Begin begins a transaction, and returns the associated transaction id.
// Subsequent statements can access the connection through the transaction id.
func (axp *TxPool) Begin(ctx context.Context) int64 {
	poolCtx := ctx
	if deadline, ok := ctx.Deadline(); ok {
		var cancel func()
		poolCtx, cancel = context.WithDeadline(ctx, deadline.Add(-10*time.Millisecond))
		defer cancel()
	}
	conn, err := axp.pool.Get(poolCtx)
	if err != nil {
		switch err {
		case ErrConnPoolClosed:
			panic(err)
		case pools.ErrTimeout:
			axp.LogActive()
			panic(NewTabletError(ErrTxPoolFull, vtrpcpb.ErrorCode_RESOURCE_EXHAUSTED, "Transaction pool connection limit exceeded"))
		}
		panic(NewTabletErrorSQL(ErrFatal, vtrpcpb.ErrorCode_INTERNAL_ERROR, err))
	}
	if _, err := conn.Exec(ctx, "begin", 1, false); err != nil {
		conn.Recycle()
		panic(NewTabletErrorSQL(ErrFail, vtrpcpb.ErrorCode_UNKNOWN_ERROR, err))
	}
	transactionID := axp.lastID.Add(1)
	axp.activePool.Register(
		transactionID,
		newTxConnection(
			conn,
			transactionID,
			axp,
			callerid.ImmediateCallerIDFromContext(ctx),
			callerid.EffectiveCallerIDFromContext(ctx),
		),
	)
	return transactionID
}

func TestPostAndParseWithRetry(t *testing.T) {
	shortLeeway := time.Millisecond * 100
	jiffy := time.Millisecond

	tests := []struct {
		uri          string
		request      interface{}
		deadlineSecs int // -1 indicates no deadline
		expected     time.Duration
		leeway       time.Duration
		retryAfter   int // -1 indicates generate 503 with no Retry-After
		failCount    int
		errstr       string
	}{
		{"/retry", nil, -1, jiffy, shortLeeway, 0, 0, ""},
		{"/error", TestParams{RespCode: 418}, 2, jiffy, shortLeeway, 0, 0, "teapot"},
		{"/short%", nil, 2, 2 * time.Second, shortLeeway, 0, 0, "deadline exceeded"},
		{"/retry", nil, -1, 7 * time.Second, shortLeeway, -1, 3, ""},
		{"/retry", nil, 6, 5 * time.Second, shortLeeway, 5, 1, ""},
		{"/retry", nil, 5, 5 * time.Second, shortLeeway, 10, 1, "deadline exceeded"},
		{"/retry", nil, 10, 5 * time.Second, shortLeeway, 1, 5, ""},
		{"/retry", nil, 1, 10 * jiffy, shortLeeway, 0, 10, ""},
		{"/retry-rfc1123", nil, -1, 2 * time.Second, 1 * time.Second, 2, 1, ""},
	}
	for _, test := range tests {
		ts := MockServer(t, test.failCount, test.retryAfter)
		defer ts.Close()

		logClient, err := New(ts.URL, &http.Client{}, Options{})
		if err != nil {
			t.Fatal(err)
		}
		ctx := context.Background()
		if test.deadlineSecs >= 0 {
			ctx, _ = context.WithDeadline(context.Background(), time.Now().Add(time.Duration(test.deadlineSecs)*time.Second))
		}

		var result TestStruct
		started := time.Now()
		httpRsp, err := logClient.PostAndParseWithRetry(ctx, test.uri, test.request, &result)
		took := time.Since(started)

		if math.Abs(float64(took-test.expected)) > float64(test.leeway) {
			t.Errorf("PostAndParseWithRetry() took %s; want ~%s", took, test.expected)
		}
		if test.errstr != "" {
			if err == nil {
				t.Errorf("PostAndParseWithRetry()=%+v,nil; want error %q", result, test.errstr)
			} else if !strings.Contains(err.Error(), test.errstr) {
				t.Errorf("PostAndParseWithRetry()=nil,%q; want error %q", err.Error(), test.errstr)
			}
			continue
		}
		if err != nil {
			t.Errorf("PostAndParseWithRetry()=nil,%q; want no error", err.Error())
		} else if httpRsp.StatusCode != http.StatusOK {
			t.Errorf("PostAndParseWithRetry() got status %d; want OK(404)", httpRsp.StatusCode)
		}
	}
}

func TestDBConnStream(t *testing.T) {
	db := fakesqldb.Register()
	testUtils := newTestUtils()
	sql := "select * from test_table limit 1000"
	expectedResult := &mproto.QueryResult{
		RowsAffected: 0,
		Rows: [][]sqltypes.Value{
			[]sqltypes.Value{sqltypes.MakeString([]byte("123"))},
		},
	}
	db.AddQuery(sql, expectedResult)
	connPool := testUtils.newConnPool()
	appParams := &sqldb.ConnParams{Engine: db.Name}
	dbaParams := &sqldb.ConnParams{Engine: db.Name}
	connPool.Open(appParams, dbaParams)
	defer connPool.Close()
	ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(10*time.Second))
	defer cancel()
	queryServiceStats := NewQueryServiceStats("", false)
	dbConn, err := NewDBConn(connPool, appParams, dbaParams, queryServiceStats)
	defer dbConn.Close()
	var result mproto.QueryResult
	err = dbConn.Stream(
		ctx, sql, func(r *mproto.QueryResult) error {
			result = *r
			return nil
		}, 10)
	if err != nil {
		t.Fatalf("should not get an error, err: %v", err)
	}
	testUtils.checkEqual(t, expectedResult, &result)
}

func resourceCloudFlareRecordDelete(d *schema.ResourceData, meta interface{}) error {
	var (
		client = meta.(*cloudflare.Client)
		domain = d.Get("domain").(string)
		rName  = strings.Join([]string{d.Get("name").(string), domain}, ".")
	)

	zone, err := retrieveZone(client, domain)
	if err != nil {
		return err
	}

	record, err := retrieveRecord(client, zone, rName)
	if err != nil {
		return err
	}

	log.Printf("[INFO] Deleting CloudFlare Record: %s, %s", domain, d.Id())

	ctx, _ := context.WithDeadline(context.Background(), time.Now().Add(time.Second*30))

	err = client.Records.Delete(ctx, zone.ID, record.ID)
	if err != nil {
		return fmt.Errorf("Error deleting CloudFlare Record: %s", err)
	}

	return nil
}

// ExecuteEntityIds is the RPC version of vtgateservice.VTGateService method
func (vtg *VTGateP3) ExecuteEntityIds(ctx context.Context, request *pb.ExecuteEntityIdsRequest, response *pb.ExecuteEntityIdsResponse) (err error) {
	defer vtg.server.HandlePanic(&err)
	ctx, cancel := context.WithDeadline(ctx, time.Now().Add(*rpcTimeout))
	defer cancel()
	ctx = callerid.NewContext(ctx,
		request.CallerId,
		callerid.NewImmediateCallerID("bsonp3 client"))
	reply := new(proto.QueryResult)
	executeErr := vtg.server.ExecuteEntityIds(ctx,
		string(request.Query.Sql),
		tproto.Proto3ToBindVariables(request.Query.BindVariables),
		request.Keyspace,
		request.EntityColumnName,
		request.EntityKeyspaceIds,
		request.TabletType,
		proto.ProtoToSession(request.Session),
		request.NotInTransaction,
		reply)
	if executeErr == nil {
		response.Error = vtgate.RPCErrorToVtRPCError(reply.Err)
		response.Result = mproto.QueryResultToProto3(reply.Result)
		response.Session = proto.SessionToProto(reply.Session)
	}
	return vterrors.ToJSONError(executeErr)
}

func retrieveRecord(
	client *cloudflare.Client,
	zone *cloudflare.Zone,
	name string,
) (*cloudflare.Record, error) {
	ctx, _ := context.WithDeadline(context.Background(), time.Now().Add(time.Second*30))

	rs, err := client.Records.List(ctx, zone.ID)
	if err != nil {
		return nil, fmt.Errorf("Unable to retrieve records for (%s): %s", zone.Name, err)
	}

	var record *cloudflare.Record

	for _, r := range rs {
		if r.Name == name {
			record = r
		}
	}
	if record == nil {
		return nil, fmt.Errorf("Unable to find Cloudflare record %s", name)
	}

	return record, nil
}

// ExecuteKeyspaceIds is the RPC version of vtgateservice.VTGateService method
func (vtg *VTGateP3) ExecuteKeyspaceIds(ctx context.Context, request *pb.ExecuteKeyspaceIdsRequest, response *pb.ExecuteKeyspaceIdsResponse) (err error) {
	defer vtg.server.HandlePanic(&err)
	ctx, cancel := context.WithDeadline(ctx, time.Now().Add(*rpcTimeout))
	defer cancel()
	ctx = callerid.NewContext(ctx,
		request.CallerId,
		callerid.NewImmediateCallerID("gorpc client"))
	reply := &proto.QueryResult{}
	vtgErr := vtg.server.ExecuteKeyspaceIds(ctx,
		string(request.Query.Sql),
		tproto.Proto3ToBindVariables(request.Query.BindVariables),
		request.Keyspace,
		request.KeyspaceIds,
		request.TabletType,
		proto.ProtoToSession(request.Session),
		request.NotInTransaction,
		reply)
	response.Error = vtgate.VtGateErrorToVtRPCError(vtgErr, reply.Error)
	response.Result = mproto.QueryResultToProto3(reply.Result)
	response.Session = proto.SessionToProto(reply.Session)
	if *vtgate.RPCErrorOnlyInReply {
		return nil
	}
	return vtgErr
}

// ContextTimeoutFailFast sets a shorter timeout in the context of interactive requests and allows
// the previous middleware to log or answer request timeouts. Be sure you process the
// "request timeout" error or it will panic when finishing the middleware stack. It is not activated
// by default; but you can clear the middleware stack and put it instead of ContextTimeout().
func ContextTimeoutFailFast(ctx context.Context, w http.ResponseWriter, r *http.Request, next NextMiddlewareFn) error {
	secs := 15 * time.Second
	if r.Header.Get("X-AppEngine-QueueName") != "" || r.Header.Get("X-AppEngine-Cron") != "" {
		secs = (9*60 + 55) * time.Second
	}

	ctx, cancel := context.WithDeadline(ctx, time.Now().Add(secs))

	done := make(chan error, 1)
	go func() {
		done <- next(ctx)
	}()

	select {
	case err := <-done:
		cancel()
		return err

	case <-time.After(secs):
		cancel()
		return errors.New("request timeout")
	}

	panic("should not reach here")
}

func testAccCheckCloudFlareRecordExists(n string, record *cloudflare.Record) resource.TestCheckFunc {
	return func(s *terraform.State) error {
		rs, ok := s.RootModule().Resources[n]
		if !ok {
			return fmt.Errorf("Not found: %s", n)
		}

		if rs.Primary.ID == "" {
			return fmt.Errorf("No Record ID is set")
		}

		var (
			client = testAccProvider.Meta().(*cloudflare.Client)
			ctx, _ = context.WithDeadline(context.Background(), time.Now().Add(time.Second*30))
		)

		foundRecord, err := client.Records.Details(ctx, rs.Primary.Attributes["zone_id"], rs.Primary.ID)
		if err != nil {
			return err
		}

		if foundRecord.ID != rs.Primary.ID {
			return fmt.Errorf("Record not found")
		}

		*record = *foundRecord

		return nil
	}
}

// WithDeadline updates contest with deadline adjusted to be no later then d.
// If deadline is later then already set deadline, semantically nothing changes.
// Context Done channel is closed when deadline expires, when returned cancel
// function is returned or when parents Done channel is closed, whichever happens
// first.
//
// Canceling context releases resources associated with it, so code should call
// cancel as soon as operation running in this Context completes.
func (c *Context) WithDeadline(deadline time.Time) (cancel context.CancelFunc) {
	c.mu.Lock()
	defer c.mu.Unlock()
	deadlineContext, cancelFunc := context.WithDeadline(c.netCtx, deadline)
	c.netCtx = deadlineContext
	return cancelFunc
}

// Begin is the RPC version of vtgateservice.VTGateService method
func (vtg *VTGate) Begin(ctx context.Context, noInput *rpc.Unused, outSession *pbg.Session) (err error) {
	defer vtg.server.HandlePanic(&err)
	ctx, cancel := context.WithDeadline(ctx, time.Now().Add(*rpcTimeout))
	defer cancel()
	err = vtg.server.Begin(ctx, outSession)
	outSession = sessionToRPC(outSession)
	return err
}

func (client *BfTrderClient) CancleOrder(req *BfCancelOrderReq) {
	ctx := context.Background()
	ctx = metadata.NewContext(ctx, metadata.Pairs("clientid", clientId_))
	ctx, cancel := context.WithDeadline(ctx, time.Now().Add(deadline*time.Second))
	defer cancel()

	client.Gateway.CancelOrder(ctx, req)
}

func (client *BfTrderClient) DeleteTick(req *BfDeleteTickReq) {
	ctx := context.Background()
	ctx = metadata.NewContext(ctx, metadata.Pairs("clientid", clientId_))
	ctx, cancel := context.WithDeadline(ctx, time.Now().Add(deadline*time.Second))
	defer cancel()

	client.Datafeed.DeleteTick(ctx, req)
}

//===datafeed api===
func (client *BfTrderClient) InsertContract(req *BfContractData) {
	ctx := context.Background()
	ctx = metadata.NewContext(ctx, metadata.Pairs("clientid", clientId_))
	ctx, cancel := context.WithDeadline(ctx, time.Now().Add(deadline*time.Second))
	defer cancel()

	client.Datafeed.InsertContract(ctx, req)
}

func (client *BfTrderClient) QueryOrders() {
	ctx := context.Background()
	ctx = metadata.NewContext(ctx, metadata.Pairs("clientid", clientId_))
	ctx, cancel := context.WithDeadline(ctx, time.Now().Add(deadline*time.Second))
	defer cancel()

	client.Gateway.QueryOrders(ctx, &BfVoid{})
}

func main() {
	defer glog.Flush()
	flag.Parse()

	if *outputFile == "" {
		glog.Fatal("--output flag cannot be empty")
	}

	c := crawl.New(
		crawl.WithConcurrency(1),
		crawl.WithQueue(crawl.NewQueue(1000)),
	)

	spider := &spider{c: c, results: make(chan *userAgent, 10000)}
	c.Register("list", spider.parseList)
	c.Register("user-agents", spider.parseUserAgents)

	ctx := context.Background()
	ctx, cancel := context.WithDeadline(ctx, time.Now().Add(time.Second*30))
	defer cancel()

	if err := c.Schedule(ctx, &crawl.Request{
		URL:       "http://www.useragentstring.com/pages/useragentstring.php",
		Callbacks: crawl.Callbacks("list"),
	}); err != nil {
		glog.Fatal(err)
	}

	glog.Info("Starting crawl")

	go func() {
		for err := range c.Errors() {
			glog.Infof("Crawl error: %v", err)
		}
	}()

	go c.Start()

	f, err := os.Create(*outputFile)
	if err != nil {
		glog.Fatal(err)
	}
	defer f.Close()

	var results []*userAgent
	for result := range spider.results {
		results = append(results, result)
	}

	b, err := json.MarshalIndent(results, "", "  ")
	if err != nil {
		glog.Fatal(err)
	}
	if _, err := f.Write(b); err != nil {
		glog.Fatal(err)
	}
	glog.Infof("Done (%d user agents)", len(results))
}

func main() {
	flag.Parse()

	exip, err := metadata.ExternalIP()
	if err != nil {
		log.Fatal(err)
	}

	client := cloudflare.New(&cloudflare.Options{
		Email: *user,
		Key:   *key,
	})

	ctx := context.Background()
	ctx, _ = context.WithDeadline(ctx, time.Now().Add(time.Second*30))

	zones, err := client.Zones.List(ctx)
	if err != nil {
		log.Fatal(err)
	} else if len(zones) == 0 {
		log.Fatal("No zones were found")
	} else if len(zones) != 1 {
		log.Fatal("More than one zone found?")
	}

	if zones[0].Name != "nella.org" {
		log.Fatal("not nella.org?")
	}

	records, err := client.Records.List(ctx, zones[0].ID)
	if err != nil {
		log.Fatal(err)
	}

	// remove all existing ns.nella.org records
	for _, record := range records {
		if record.Name == "ns.nella.org" {
			log.Print("delete ", record.Content)
			err = client.Records.Delete(ctx, record.ZoneID, record.ID)
			if err != nil {
				log.Fatal("delete: ", err)
			}
		}
	}

	rec := &cloudflare.Record{
		Type:    "A",
		Name:    "ns.nella.org",
		Content: exip,
		TTL:     120,
		ZoneID:  zones[0].ID,
	}
	log.Print("create: ", exip)
	err = client.Records.Create(ctx, rec)
	if err != nil {
		log.Fatal("create: ", err)
	}
}

func TestContextCancelOnRetry(t *testing.T) {
	callCnt := int32(0)
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		cnt := atomic.AddInt32(&callCnt, 1)
		switch cnt {
		case 1:
			w.WriteHeader(500)
		default:
			time.Sleep(2 * time.Second)
			fmt.Fprint(w, r.URL.Path)
		}
	}))
	defer srv.Close()

	// cancel while waiting on retry response
	tr := NewTransport(nil, RetryAll(RetryMaxRetries(1), RetryStatusInterval(500, 600)), ConstDelay(0))
	c := &http.Client{
		Transport: tr,
	}

	ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(time.Second))
	defer cancelFn()

	res, err := ctxhttp.Get(ctx, c, srv.URL+"/test")
	require.Nil(t, res)

	assert.Equal(t, context.DeadlineExceeded, err)
	assert.Equal(t, int32(2), atomic.LoadInt32(&callCnt))

	// cancel while waiting on delay
	atomic.StoreInt32(&callCnt, 0)
	tr = NewTransport(nil, RetryAll(RetryMaxRetries(1), RetryStatusInterval(500, 600)), ConstDelay(2*time.Second))
	c = &http.Client{
		Transport: tr,
	}

	ctx, cancelFn = context.WithDeadline(context.Background(), time.Now().Add(time.Second))
	defer cancelFn()

	res, err = ctxhttp.Get(ctx, c, srv.URL+"/test")
	require.Nil(t, res)

	assert.Equal(t, context.DeadlineExceeded, err)
	assert.Equal(t, int32(1), atomic.LoadInt32(&callCnt))
}