// Sends an AppendEntries RPC to a peer.
func (t *HTTPTransporter) SendAppendEntriesRequest(server raft.Server, peer *raft.Peer, req *raft.AppendEntriesRequest) *raft.AppendEntriesResponse {
	var b bytes.Buffer
	if _, err := req.Encode(&b); err != nil {
		//log.Println("transporter.ae.encoding.error:", err)
		return nil
	}

	url := joinPath(peer.ConnectionString, t.AppendEntriesPath())
	//log.Println(server.Name(), "POST", url)

	//oldTimeout := t.Transport.ResponseHeaderTimeout
	t.Transport.ResponseHeaderTimeout = server.ElectionTimeout()
	httpResp, err := t.httpClient.Post(url, "application/protobuf", &b)
	if httpResp == nil || err != nil {
		//log.Println("transporter.ae.response.error:", err)
		return nil
	}
	//t.Transport.ResponseHeaderTimeout = oldTimeout
	defer httpResp.Body.Close()

	resp := &raft.AppendEntriesResponse{}
	if _, err = resp.Decode(httpResp.Body); err != nil && err != io.EOF {
		//log.Println("transporter.ae.decoding.error:", err)
		return nil
	}

	return resp
}

func (c *AddCallbackCommand) Apply(server raft.Server) (interface{}, error) {
	reg := server.Context().(registry.Registry)
	err := reg.AddCallback(c.Service, c.Callback)
	if err == nil {
		log.Println("Added Callback:", c.Service, c.Callback)
	}
	return c.Service, err
}

//???
func (c *MaxVolumeIdCommand) Apply(server raft.Server) (interface{}, error) {
	topo := server.Context().(*Topology)
	before := topo.GetMaxVolumeId()
	topo.UpAdjustMaxVolumeId(c.MaxVolumeId)

	glog.V(4).Infoln("max volume id", before, "==>", topo.GetMaxVolumeId())

	return nil, nil
}

// Writes a value to a key.
func (c *QueryCommand) Apply(server raft.Server) (interface{}, error) {
	db := server.Context().(*sql.SQL)

	output, err := db.CacheExec(c.Create, c.Name, c.FriendCount, c.FavoriteWord)
	if err != nil {
		return nil, err
	}

	formatted := fmt.Sprintf("SequenceNumber: %d\n%s", output.SequenceNumber, output.Output)

	return []byte(formatted), nil
}

func (c *CreateShardsCommand) Apply(server raft.Server) (interface{}, error) {
	config := server.Context().(*cluster.ClusterConfiguration)
	createdShards, err := config.AddShards(c.Shards)
	if err != nil {
		return nil, err
	}
	createdShardData := make([]*cluster.NewShardData, 0)
	for _, s := range createdShards {
		createdShardData = append(createdShardData, s.ToNewShardData())
	}
	return createdShardData, nil
}

func (c *SaveSubscriptionCommand) Apply(server raft.Server) (interface{}, error) {
	config := server.Context().(*cluster.ClusterConfiguration)
	createdSubscription, err := config.AddSubscription(c.Subscription)
	fmt.Println("we are here in command.go's Apply function")
	if err != nil {
		return nil, err
	}
	//createdSubscriptionData := make(*cluster.Subscription, 0)
	//for _, s := range createdSubscription {
	//createdSubscriptionData = append(createdSubscriptionData, s.ToNewSubscriptionData())
	//}
	//return createdSubscriptionData, nil
	return createdSubscription, nil
	return nil, nil
}

// Execute an SQL statement
func (c *WriteCommand) Apply(server raft.Server) (interface{}, error) {
	sql := server.Context().(*sql.SQL)
	job := sql.Execute(c.Query)

	go func() {
		output := []byte(<-job.OutChan)
		seq := job.Seq

		formatted := fmt.Sprintf("SequenceNumber: %d\n%s",
			seq, output)

		sql.Respond(c.Id, []byte(formatted))
	}()

	return []byte(""), nil
}

// Handles incoming SnapshotRecovery requests.
func (t *HTTPTransporter) snapshotRecoveryHandler(server raft.Server) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		//log.Println(server.Name(), "RECV /snapshotRecovery")

		req := &raft.SnapshotRecoveryRequest{}
		if _, err := req.Decode(r.Body); err != nil {
			http.Error(w, "", http.StatusBadRequest)
			return
		}

		resp := server.SnapshotRecoveryRequest(req)
		if _, err := resp.Encode(w); err != nil {
			http.Error(w, "", http.StatusInternalServerError)
			return
		}
	}
}

// Writes a value to a key.
func (c *BatchCommand) Apply(server raft.Server) (interface{}, error) {
	//log.Printf(("Applying Batch Size: %v", len(c.Queries))
	db := server.Context().(*DB)

	for _, output := range db.ExecuteBatch(c.Queries) {
		db.notify <- output
	}

	// for _, query := range c.Queries {
	// 	output, err := db.Execute(*query)

	// 	if err != nil {
	// 		log.Fatal("Error applying command")
	// 		return nil, err
	// 	}

	// 	db.notify <- output

	// }
	return nil, nil
}

// Writes a value to a key.
func (c *SqlCommand) Apply(server raft.Server) (interface{}, error) {
	db := server.Context().(*sql.SQL)
	output, err := db.Execute("tehrafts", inflate(c.Query))

	if err != nil {
		var msg string
		if output != nil && len(output.Stderr) > 0 {
			template := `Error executing %#v (%s)

SQLite error: %s`
			msg = fmt.Sprintf(template, c.Query, err.Error(), util.FmtOutput(output.Stderr))
		} else {
			msg = err.Error()
		}

		return nil, errors.New(msg)
	}

	formatted := fmt.Sprintf("SequenceNumber: %d\n%s",
		output.SequenceNumber, output.Stdout)
	return []byte(formatted), nil
}

// Apply implements goraft Command interface's Apply function
// It puts a key-value pair in KVstore
func (pcmd *putCommand) Apply(server raft.Server) (interface{}, error) {
	if server.Name() == server.Leader() {
		return nil, nil
	}
	kvs := server.Context().(KVstore)
	err := kvs.Put(pcmd.Key, pcmd.Val)
	return nil, err
}

func (c *WriteCommand) Apply(server raft.Server) (interface{}, error) {
	db := server.Context().(*DB)
	db.Put(c.Key, c.Value)
	return nil, nil
}

func (c *UpdateServerStateCommand) Apply(server raft.Server) (interface{}, error) {
	config := server.Context().(*ClusterConfiguration)
	err := config.UpdateServerState(c.ServerId, c.State)
	return nil, err
}

func (c *AddPotentialServerCommand) Apply(server raft.Server) (interface{}, error) {
	config := server.Context().(*cluster.ClusterConfiguration)
	config.AddPotentialServer(c.Server)
	return nil, nil
}

// Writes a value to a key.
func (c *RemoveTmpDataCommand) Apply(server raft.Server) (interface{}, error) {
	fs := server.Context().(*db.Fs)
	//log.Printf("RemoveTempDataIp: %v\n", c.Key)
	fs.RemoveTempData(c.Key)
	return nil, nil
}