// usageString returns a description of the amount of space taken up by a body
// with the given contents and a bool indicating overflow.
func (draft *Draft) usageString() (string, bool) {
	var replyToId *uint64
	if draft.inReplyTo != 0 {
		replyToId = proto.Uint64(1)
	}
	var dhPub [32]byte

	msg := &pond.Message{
		Id:               proto.Uint64(0),
		Time:             proto.Int64(1 << 62),
		Body:             []byte(draft.body),
		BodyEncoding:     pond.Message_RAW.Enum(),
		InReplyTo:        replyToId,
		MyNextDh:         dhPub[:],
		Files:            draft.attachments,
		DetachedFiles:    draft.detachments,
		SupportedVersion: proto.Int32(protoVersion),
	}

	serialized, err := proto.Marshal(msg)
	if err != nil {
		panic("error while serialising candidate Message: " + err.Error())
	}

	s := fmt.Sprintf("%d of %d bytes", len(serialized), pond.MaxSerializedMessage)
	return s, len(serialized) > pond.MaxSerializedMessage
}

// Update the datastore with the user's current state.
func (server *Server) UpdateFrozenUser(client *Client, state *mumbleproto.UserState) {
	// Full sync If there's no userstate messgae provided, or if there is one, and
	// it includes a registration operation.
	user := client.user
	nanos := time.Now().Unix()
	if state == nil || state.UserId != nil {
		fu, err := user.Freeze()
		if err != nil {
			server.Fatal(err)
		}
		fu.LastActive = proto.Uint64(uint64(nanos))
		err = server.freezelog.Put(fu)
		if err != nil {
			server.Fatal(err)
		}
	} else {
		fu := &freezer.User{}
		fu.Id = proto.Uint32(user.Id)
		if state.ChannelId != nil {
			fu.LastChannelId = proto.Uint32(uint32(client.Channel.Id))
		}
		if state.TextureHash != nil {
			fu.TextureBlob = proto.String(user.TextureBlob)
		}
		if state.CommentHash != nil {
			fu.CommentBlob = proto.String(user.CommentBlob)
		}
		fu.LastActive = proto.Uint64(uint64(nanos))
		err := server.freezelog.Put(fu)
		if err != nil {
			server.Fatal(err)
		}
	}
	server.numLogOps += 1
}

// Encodes the SnapshotRecoveryRequest to a buffer. Returns the number of bytes
// written and any error that may have occurred.
func (req *SnapshotRecoveryRequest) encode(w io.Writer) (int, error) {

	protoPeers := make([]*protobuf.ProtoSnapshotRecoveryRequest_ProtoPeer, len(req.Peers))

	for i, peer := range req.Peers {
		protoPeers[i] = &protobuf.ProtoSnapshotRecoveryRequest_ProtoPeer{
			Name:             proto.String(peer.Name),
			ConnectionString: proto.String(peer.ConnectionString),
		}
	}

	pb := &protobuf.ProtoSnapshotRecoveryRequest{
		LeaderName: proto.String(req.LeaderName),
		LastIndex:  proto.Uint64(req.LastIndex),
		LastTerm:   proto.Uint64(req.LastTerm),
		Peers:      protoPeers,
		State:      req.State,
	}
	p, err := proto.Marshal(pb)
	if err != nil {
		return -1, err
	}

	return w.Write(p)
}

func (self *InterfaceForProject) PushMessage(interfaceCMD *business.Command) (resultPacket Packet, err error) {
	//1.从interfaceCMD 中解析出所需要的信息
	//2.处理，并且得到结果
	//3.把结果写入CMD，并且做成Packet 返回。

	//1.解析...................................
	if interfaceCMD.GetUid() == nil {
		feedbackCMD.Result = proto.Uint32(2)
		feedbackCMD.ResultDescription = proto.String("no uid")
		resultPacket = *MakeInterfacePacket(feedbackCMD, 102)
		return resultPacket, err
	}
	senderUid := interfaceCMD.GetUid()[0]
	dialogID := interfaceCMD.GetGroupid()
	if dialogID == 0 {
		feedbackCMD.Result = proto.Uint32(2)
		feedbackCMD.ResultDescription = proto.String("no dialogid")
		resultPacket = *MakeInterfacePacket(feedbackCMD, 102)
		return resultPacket, err
	}
	if interfaceCMD.GetMessage() == nil {
		feedbackCMD.Result = proto.Uint32(2)
		feedbackCMD.ResultDescription = proto.String("no message")
		resultPacket = *MakeInterfacePacket(feedbackCMD, 102)
		return resultPacket, err
	}
	message := interfaceCMD.GetMessage()[0]
	random := GenRandom()
	var (
		requestMess = &protocol.Message{
			Data:   proto.String(message),
			Random: proto.Uint32(random),
		}
		requestCMD = &protocol.Command{
			Sender:   proto.Uint64(senderUid), //sender's uid
			Receiver: proto.Uint64(dialogID),  //0=server
			Message:  requestMess,
		}
	)
	pushMessagePacket := *MakePacket(requestCMD, 2)
	_, fl := SessionTable[senderUid]
	if fl == false {
		//fmt.Println("uid no in")
		feedbackCMD.Result = proto.Uint32(1)
		feedbackCMD.ResultDescription = proto.String("sender isn't online")
		resultPacket = *MakeInterfacePacket(feedbackCMD, 102)
		return resultPacket, nil
	}

	SessionTable[senderUid].incoming <- pushMessagePacket
	feedbackCMD.Result = proto.Uint32(0)
	feedbackCMD.ResultDescription = proto.String("success")
	resultPacket = *MakeInterfacePacket(feedbackCMD, 102)
	return resultPacket, nil
	//2.处理...................................
	//2.1 从dialogID （act_dialog表---dialog表）       中得到receiverUid,若不为senderUid，则转发，或者存离线
	//2.2 查找receiverUid失败，则返回要求服务器先建群
	//3.打包...................................
}

func (h *Harvester) Harvest(output chan *FileEvent) {
	// TODO(sissel): Read the file
	// TODO(sissel): Emit FileEvent for each line to 'output'
	// TODO(sissel): Handle rotation
	// TODO(sissel): Sleep when there's nothing to do
	// TODO(sissel): Quit if we think the file is dead (file dev/inode changed, no data in X seconds)

	fmt.Printf("Starting harvester: %s\n", h.Path)

	file := h.open()
	defer file.Close()
	//info, _ := file.Stat()

	// TODO(sissel): Ask the registrar for the start position?
	// TODO(sissel): record the current file inode/device/etc

	var line uint64 = 0 // Ask registrar about the line number

	// get current offset in file
	offset, _ := file.Seek(0, os.SEEK_CUR)

	// TODO(sissel): Make the buffer size tunable at start-time
	reader := bufio.NewReaderSize(file, 16<<10) // 16kb buffer by default

	var read_timeout = 10 * time.Second
	last_read_time := time.Now()
	for {
		text, err := h.readline(reader, read_timeout)

		if err != nil {
			if err == io.EOF {
				// timed out waiting for data, got eof.
				// TODO(sissel): Check to see if the file was truncated
				// TODO(sissel): if last_read_time was more than 24 hours ago
				if age := time.Since(last_read_time); age > (24 * time.Hour) {
					// This file is idle for more than 24 hours. Give up and stop harvesting.
					fmt.Printf("Stopping harvest of %s; last change was %d seconds ago\n", h.Path, age.Seconds())
					return
				}
				continue
			} else {
				fmt.Printf("Unexpected state reading from %s; error: %s\n", h.Path, err)
				return
			}
		}
		last_read_time = time.Now()

		line++
		event := &FileEvent{
			Source: proto.String(h.Path),
			Offset: proto.Uint64(uint64(offset)),
			Line:   proto.Uint64(line),
			Text:   text,
		}
		offset += int64(len(*event.Text)) + 1 // +1 because of the line terminator

		output <- event // ship the new event downstream
	} /* forever */
}

func fillResult(result *subprocess.SubprocessResult, response *contester_proto.LocalExecutionResult) {
	if result.TotalProcesses > 0 {
		response.TotalProcesses = proto.Uint64(result.TotalProcesses)
	}
	response.ReturnCode = proto.Uint32(result.ExitCode)
	response.Flags = parseSuccessCode(result.SuccessCode)
	response.Time = parseTime(result)
	response.Memory = proto.Uint64(result.PeakMemory)
	response.StdOut, _ = contester_proto.NewBlob(result.Output)
	response.StdErr, _ = contester_proto.NewBlob(result.Error)
}

// Encodes the SnapshotRequest to a buffer. Returns the number of bytes
// written and any error that may have occurred.
func (req *SnapshotRequest) Encode(w io.Writer) (int, error) {
	pb := &protobuf.ProtoSnapshotRequest{
		LeaderName: proto.String(req.LeaderName),
		LastIndex:  proto.Uint64(req.LastIndex),
		LastTerm:   proto.Uint64(req.LastTerm),
	}
	p, err := proto.Marshal(pb)
	if err != nil {
		return -1, err
	}

	return w.Write(p)
}

// Encodes the SnapshotRecoveryResponse to a buffer. Returns the number of bytes
// written and any error that may have occurred.
func (req *SnapshotRecoveryResponse) encode(w io.Writer) (int, error) {
	pb := &protobuf.ProtoSnapshotRecoveryResponse{
		Term:        proto.Uint64(req.Term),
		Success:     proto.Bool(req.Success),
		CommitIndex: proto.Uint64(req.CommitIndex),
	}
	p, err := proto.Marshal(pb)
	if err != nil {
		return -1, err
	}

	return w.Write(p)
}

// Encodes the AppendEntriesResponse to a buffer. Returns the number of bytes
// written and any error that may have occurred.
func (resp *AppendEntriesResponse) encode(w io.Writer) (int, error) {
	pb := &protobuf.ProtoAppendEntriesResponse{
		Term:        proto.Uint64(resp.Term),
		Index:       proto.Uint64(resp.Index),
		CommitIndex: proto.Uint64(resp.CommitIndex),
		Success:     proto.Bool(resp.Success),
	}
	p, err := proto.Marshal(pb)
	if err != nil {
		return -1, err
	}

	return w.Write(p)
}

// Encodes the RequestVoteRequest to a buffer. Returns the number of bytes
// written and any error that may have occurred.
func (req *RequestVoteRequest) Encode(w io.Writer) (int, error) {
	pb := &protobuf.ProtoRequestVoteRequest{
		Term:          proto.Uint64(req.Term),
		LastLogIndex:  proto.Uint64(req.LastLogIndex),
		LastLogTerm:   proto.Uint64(req.LastLogTerm),
		CandidateName: proto.String(req.CandidateName),
	}
	p, err := proto.Marshal(pb)
	if err != nil {
		return -1, err
	}

	return w.Write(p)
}

func packBlock(nrows uint64, data []byte) (*block, error) {
	compressed, err := lzmaCompress(int64(len(data)), data)
	if err != nil {
		return nil, err
	}

	header := cbs_proto.Header{
		NumRows:             proto.Uint64(nrows),
		BlockSize:           proto.Uint64(uint64(len(data))),
		CompressedBlockSize: proto.Uint64(uint64(len(compressed))),
	}

	return &block{&header, compressed}, nil
}

func RPCAuthenticateWithKeyMessage(conn net.Conn, connection_data *structs.ConnData, packet_data *structs.PacketData) error {
	// Unmarshal the data
	msg := new(protocol.AuthenticateWithKeyMessage)
	err := proto.Unmarshal(packet_data.Content, msg)
	if err != nil {
		return err
	}

	// Generate a new connection-id based npid
	npid := structs.IdToNpid(connection_data.ConnectionId)

	// Fill in the connection data
	connection_data.Authenticated = true
	connection_data.IsServer = true
	connection_data.Token = ""
	connection_data.Npid = npid

	// Add connection to the storage
	storage.SetServerConnection(npid, connection_data)

	// Reply with the data
	return reply.Reply(conn, packet_data.Header.Id, &protocol.AuthenticateResultMessage{
		Result:       proto.Int32(0),
		Npid:         proto.Uint64(npid),
		SessionToken: []byte(""),
	})
}

func (sc *ServerClient) dispatch() {
	for p := range sc.in {
		if p.GetEnsureArrival() != 0 {
			sc.Send(&packet.Packet{
				ArrivalNotice: &packet.Packet_ArrivalNotice{
					PacketID: proto.Uint64(p.GetEnsureArrival()),
				},
			}, false)
		} else if an := p.GetArrivalNotice().GetPacketID(); an != 0 {
			sc.server.arrivalNotice(sc.addr, an)
			continue
		}

		if t := p.GetTestingPacket(); t != nil {
			switch t.GetType() {
			case packet.Packet_TestingPacket_Push:
			case packet.Packet_TestingPacket_Request:
				sc.Send(&packet.Packet{
					TestingPacket: &packet.Packet_TestingPacket{
						Type: packet.Packet_TestingPacket_Response.Enum(),
					},
				}, false)

			case packet.Packet_TestingPacket_Response:
			}
			continue
		}
		log.Println(sc.addr, p)
	}
}

func (c *clientCodec) WriteRequest(rpcreq *rpc.Request, param interface{}) error {
	rr := *rpcreq
	req := &Request{}

	c.mutex.Lock()
	req.Id = proto.Uint64(c.next)
	c.next++
	c.pending[*req.Id] = &rr
	c.mutex.Unlock()

	req.Method = proto.String(rpcreq.ServiceMethod)
	if msg, ok := param.(proto.Message); ok {
		body, err := proto.Marshal(msg)
		if err != nil {
			return err
		}
		req.Body = body
	} else {
		return fmt.Errorf("marshal request param error: %s", param)
	}

	f, err := proto.Marshal(req)
	if err != nil {
		return err
	}

	if err := write(c.w, f); err != nil {
		return err
	}

	return nil
}

func StatFile(name string, hash_it bool) (*contester_proto.FileStat, error) {
	result := &contester_proto.FileStat{}
	result.Name = &name
	info, err := os.Stat(name)
	if err != nil {
		// Handle ERROR_FILE_NOT_FOUND - return no error and nil instead of stat struct
		if IsStatErrorFileNotFound(err) {
			return nil, nil
		}

		return nil, NewError(err, "statFile", "os.Stat")
	}
	if info.IsDir() {
		result.IsDirectory = proto.Bool(true)
	} else {
		result.Size = proto.Uint64(uint64(info.Size()))
		if hash_it {
			checksum, err := HashFileString(name)
			if err != nil {
				return nil, NewError(err, "statFile", "hashFile")
			}
			result.Checksum = &checksum
		}
	}
	return result, nil
}

// TODO: Documentation
func (c *Client) Read() (*packet.Packet, error) {
	var buf [MaxPacketSize]byte
	parsed := new(packet.Packet)

	for {
		n, err := c.conn.Read(buf[:])
		if err != nil {
			return nil, err
		}

		if err = proto.Unmarshal(buf[:n], parsed); err != nil {
			return nil, err
		}

		if an := parsed.GetArrivalNotice().GetPacketID(); an != 0 {
			c.lock.Lock()
			delete(c.ensured, an)
			c.lock.Unlock()
			continue
		}

		if ensure := parsed.GetEnsureArrival(); ensure != 0 {
			c.Write(&packet.Packet{
				ArrivalNotice: &packet.Packet_ArrivalNotice{
					PacketID: proto.Uint64(ensure),
				},
			}, false)
		}

		return parsed, nil
	}

	panic("unreachable")
}

func createSubmit(db *sqlx.DB, sub *scannedSubmit, submitNo int) *tickets.Ticket_Submit {
	var result tickets.Ticket_Submit
	result.SubmitNumber = proto.Uint32(uint32(submitNo))
	if sub.Arrived > 0 {
		result.Arrived = proto.Uint64(uint64(sub.Arrived))
	}
	if result.Compiled = proto.Bool(sub.Compiled == 1); !result.GetCompiled() {
		return &result
	}
	if sub.SchoolMode != 0 {
		result.School = &tickets.Ticket_Submit_School{TestsTaken: proto.Uint32(uint32(sub.Taken)), TestsPassed: proto.Uint32(uint32(sub.Passed))}
	} else {
		var description string
		var test int64
		err := db.QueryRow("select ResultDesc.Description, Results.Test from Results, ResultDesc where "+
			"Results.UID = ? and ResultDesc.ID = Results.Result and not ResultDesc.Success order by Results.Test",
			sub.TestingID).Scan(&description, &test)
		switch {
		case err == sql.ErrNoRows:
			if sub.Passed != 0 && sub.Passed == sub.Taken {
				result.Acm = &tickets.Ticket_Submit_ACM{Result: proto.String("ACCEPTED")}
			}
		case err != nil:
			log.Fatal(err)
			return nil
		default:
			result.Acm = &tickets.Ticket_Submit_ACM{Result: &description, TestId: proto.Uint32(uint32(test))}
		}
	}
	return &result
}

func (s *summary) Write(out *dto.Metric) error {
	sum := &dto.Summary{}
	qs := make([]*dto.Quantile, 0, len(s.objectives))

	s.bufMtx.Lock()
	s.mtx.Lock()

	if len(s.hotBuf) != 0 {
		s.swapBufs(time.Now())
	}
	s.bufMtx.Unlock()

	s.flushColdBuf()
	sum.SampleCount = proto.Uint64(s.cnt)
	sum.SampleSum = proto.Float64(s.sum)

	for _, rank := range s.sortedObjectives {
		qs = append(qs, &dto.Quantile{
			Quantile: proto.Float64(rank),
			Value:    proto.Float64(s.headStream.Query(rank)),
		})
	}

	s.mtx.Unlock()

	if len(qs) > 0 {
		sort.Sort(quantSort(qs))
	}
	sum.Quantile = qs

	out.Summary = sum
	out.Label = s.labelPairs
	return nil
}

func (self *Client) SendBussiness(ziptype int32, datatype int32, data []byte) (uint64, string) {
	fun := "Client.SendBussiness"

	msgid, err := self.manager.Msgid()
	if err != nil {
		slog.Errorf("%s get msgid error:%s", fun, err)
		return 0, self.remoteaddr
	}

	buss := &pushproto.Talk{
		Type:     pushproto.Talk_BUSSINESS.Enum(),
		Msgid:    proto.Uint64(msgid),
		Ziptype:  proto.Int32(ziptype),
		Datatype: proto.Int32(datatype),
		Bussdata: data,
	}

	spb, err := proto.Marshal(buss)
	if err != nil {
		slog.Errorf("%s marshaling error: ", fun, err)
		return 0, self.remoteaddr
	}

	p := util.Packdata(spb)
	self.sendBussRetry(msgid, p)

	slog.Infof("%s client:%s send msgid:%d", fun, self, msgid)
	self.Send(p)

	return msgid, self.remoteaddr
}

func (db *Db) NewTransaction(cb func(b Block)) *Transaction {
	trx := &Transaction{
		Id: pb.Uint64(uint64(time.Now().UnixNano())), // TODO: put real id from nrv
	}
	cb(trx.newBlock())
	return trx
}