// Send sends a message.
// If any failures occur with specific recipients, the error will be an appengine.MultiError.
func (m *Message) Send(c context.Context) error {
	req := &pb.XmppMessageRequest{
		Jid:    m.To,
		Body:   &m.Body,
		RawXml: &m.RawXML,
	}
	if m.Type != "" && m.Type != "chat" {
		req.Type = &m.Type
	}
	if m.Sender != "" {
		req.FromJid = &m.Sender
	}
	res := &pb.XmppMessageResponse{}
	if err := internal.Call(c, "xmpp", "SendMessage", req, res); err != nil {
		return err
	}

	if len(res.Status) != len(req.Jid) {
		return fmt.Errorf("xmpp: sent message to %d JIDs, but only got %d statuses back", len(req.Jid), len(res.Status))
	}
	me, any := make(appengine.MultiError, len(req.Jid)), false
	for i, st := range res.Status {
		if st != pb.XmppMessageResponse_NO_ERROR {
			me[i] = errors.New(st.String())
			any = true
		}
	}
	if any {
		return me
	}
	return nil
}

func moreList(t *Iterator) error {
	req := &pb.ListDocumentsRequest{
		Params: &pb.ListDocumentsParams{
			IndexSpec: &t.index.spec,
		},
	}
	if t.listStartID != "" {
		req.Params.StartDocId = &t.listStartID
		req.Params.IncludeStartDoc = &t.listInclusive
	}
	if t.limit > 0 {
		req.Params.Limit = proto.Int32(int32(t.limit))
	}
	if t.idsOnly {
		req.Params.KeysOnly = &t.idsOnly
	}

	res := &pb.ListDocumentsResponse{}
	if err := internal.Call(t.c, "search", "ListDocuments", req, res); err != nil {
		return err
	}
	if res.Status == nil || res.Status.GetCode() != pb.SearchServiceError_OK {
		return fmt.Errorf("search: %s: %s", res.Status.GetCode(), res.Status.GetErrorDetail())
	}
	t.listRes = res.Document
	t.listStartID, t.listInclusive, t.more = "", false, nil
	if len(res.Document) != 0 {
		if id := res.Document[len(res.Document)-1].GetId(); id != "" {
			t.listStartID, t.more = id, moreList
		}
	}
	return nil
}

// QueueStats retrieves statistics about queues.
func QueueStats(c context.Context, queueNames []string) ([]QueueStatistics, error) {
	req := &pb.TaskQueueFetchQueueStatsRequest{
		QueueName: make([][]byte, len(queueNames)),
	}
	for i, q := range queueNames {
		if q == "" {
			q = "default"
		}
		req.QueueName[i] = []byte(q)
	}
	res := &pb.TaskQueueFetchQueueStatsResponse{}
	if err := internal.Call(c, "taskqueue", "FetchQueueStats", req, res); err != nil {
		return nil, err
	}
	qs := make([]QueueStatistics, len(res.Queuestats))
	for i, qsg := range res.Queuestats {
		qs[i] = QueueStatistics{
			Tasks: int(*qsg.NumTasks),
		}
		if eta := *qsg.OldestEtaUsec; eta > -1 {
			qs[i].OldestETA = time.Unix(0, eta*1e3)
		}
		if si := qsg.ScannerInfo; si != nil {
			qs[i].Executed1Minute = int(*si.ExecutedLastMinute)
			qs[i].InFlight = int(si.GetRequestsInFlight())
			qs[i].EnforcedRate = si.GetEnforcedRate()
		}
	}
	return qs, nil
}

func (cn *Conn) Read(b []byte) (n int, err error) {
	const maxRead = 1 << 20
	if len(b) > maxRead {
		b = b[:maxRead]
	}

	req := &pb.ReceiveRequest{
		SocketDescriptor: &cn.desc,
		DataSize:         proto.Int32(int32(len(b))),
	}
	res := &pb.ReceiveReply{}
	if !cn.readDeadline.IsZero() {
		req.TimeoutSeconds = proto.Float64(cn.readDeadline.Sub(time.Now()).Seconds())
	}
	ctx, cancel := withDeadline(cn.ctx, cn.readDeadline)
	defer cancel()
	if err := internal.Call(ctx, "remote_socket", "Receive", req, res); err != nil {
		return 0, err
	}
	if len(res.Data) == 0 {
		return 0, io.EOF
	}
	if len(res.Data) > len(b) {
		return 0, fmt.Errorf("socket: internal error: read too much data: %d > %d", len(res.Data), len(b))
	}
	return copy(b, res.Data), nil
}

func (cn *Conn) Write(b []byte) (n int, err error) {
	const lim = 1 << 20 // max per chunk

	for n < len(b) {
		chunk := b[n:]
		if len(chunk) > lim {
			chunk = chunk[:lim]
		}

		req := &pb.SendRequest{
			SocketDescriptor: &cn.desc,
			Data:             chunk,
			StreamOffset:     &cn.offset,
		}
		res := &pb.SendReply{}
		if !cn.writeDeadline.IsZero() {
			req.TimeoutSeconds = proto.Float64(cn.writeDeadline.Sub(time.Now()).Seconds())
		}
		ctx, cancel := withDeadline(cn.ctx, cn.writeDeadline)
		defer cancel()
		if err = internal.Call(ctx, "remote_socket", "Send", req, res); err != nil {
			// assume zero bytes were sent in this RPC
			break
		}
		n += int(res.GetDataSent())
		cn.offset += int64(res.GetDataSent())
	}

	return
}

// AllocateIDs returns a range of n integer IDs with the given kind and parent
// combination. kind cannot be empty; parent may be nil. The IDs in the range
// returned will not be used by the datastore's automatic ID sequence generator
// and may be used with NewKey without conflict.
//
// The range is inclusive at the low end and exclusive at the high end. In
// other words, valid intIDs x satisfy low <= x && x < high.
//
// If no error is returned, low + n == high.
func AllocateIDs(c context.Context, kind string, parent *Key, n int) (low, high int64, err error) {
	if kind == "" {
		return 0, 0, errors.New("datastore: AllocateIDs given an empty kind")
	}
	if n < 0 {
		return 0, 0, fmt.Errorf("datastore: AllocateIDs given a negative count: %d", n)
	}
	if n == 0 {
		return 0, 0, nil
	}
	req := &pb.AllocateIdsRequest{
		ModelKey: keyToProto("", NewIncompleteKey(c, kind, parent)),
		Size:     proto.Int64(int64(n)),
	}
	res := &pb.AllocateIdsResponse{}
	if err := internal.Call(c, "datastore_v3", "AllocateIds", req, res); err != nil {
		return 0, 0, err
	}
	// The protobuf is inclusive at both ends. Idiomatic Go (e.g. slices, for loops)
	// is inclusive at the low end and exclusive at the high end, so we add 1.
	low = res.GetStart()
	high = res.GetEnd() + 1
	if low+int64(n) != high {
		return 0, 0, fmt.Errorf("datastore: internal error: could not allocate %d IDs", n)
	}
	return low, high, nil
}

func (n *namespacedContext) call(ctx context.Context, service, method string, in, out proto.Message) error {
	// Apply any namespace mods.
	if mod, ok := internal.NamespaceMods[service]; ok {
		mod(in, n.namespace)
	}
	return internal.Call(ctx, service, method, in, out)
}

// Get loads the document with the given ID into dst.
//
// The ID is a human-readable ASCII string. It must be non-empty, contain no
// whitespace characters and not start with "!".
//
// dst must be a non-nil struct pointer or implement the FieldLoadSaver
// interface.
//
// ErrFieldMismatch is returned when a field is to be loaded into a different
// type than the one it was stored from, or when a field is missing or
// unexported in the destination struct. ErrFieldMismatch is only returned if
// dst is a struct pointer. It is up to the callee to decide whether this error
// is fatal, recoverable, or ignorable.
func (x *Index) Get(c context.Context, id string, dst interface{}) error {
	if id == "" || !validIndexNameOrDocID(id) {
		return fmt.Errorf("search: invalid ID %q", id)
	}
	req := &pb.ListDocumentsRequest{
		Params: &pb.ListDocumentsParams{
			IndexSpec:  &x.spec,
			StartDocId: proto.String(id),
			Limit:      proto.Int32(1),
		},
	}
	res := &pb.ListDocumentsResponse{}
	if err := internal.Call(c, "search", "ListDocuments", req, res); err != nil {
		return err
	}
	if res.Status == nil || res.Status.GetCode() != pb.SearchServiceError_OK {
		return fmt.Errorf("search: %s: %s", res.Status.GetCode(), res.Status.GetErrorDetail())
	}
	if len(res.Document) != 1 || res.Document[0].GetId() != id {
		return ErrNoSuchDocument
	}
	metadata := &DocumentMetadata{
		Rank: int(res.Document[0].GetOrderId()),
	}
	return loadDoc(dst, res.Document[0].Field, nil, metadata)
}

func lease(c context.Context, maxTasks int, queueName string, leaseTime int, groupByTag bool, tag []byte) ([]*Task, error) {
	if queueName == "" {
		queueName = "default"
	}
	req := &pb.TaskQueueQueryAndOwnTasksRequest{
		QueueName:    []byte(queueName),
		LeaseSeconds: proto.Float64(float64(leaseTime)),
		MaxTasks:     proto.Int64(int64(maxTasks)),
		GroupByTag:   proto.Bool(groupByTag),
		Tag:          tag,
	}
	res := &pb.TaskQueueQueryAndOwnTasksResponse{}
	if err := internal.Call(c, "taskqueue", "QueryAndOwnTasks", req, res); err != nil {
		return nil, err
	}
	tasks := make([]*Task, len(res.Task))
	for i, t := range res.Task {
		tasks[i] = &Task{
			Payload:    t.Body,
			Name:       string(t.TaskName),
			Method:     "PULL",
			ETA:        time.Unix(0, *t.EtaUsec*1e3),
			RetryCount: *t.RetryCount,
			Tag:        string(t.Tag),
		}
	}
	return tasks, nil
}

// KeepAlive signals that the connection is still in use.
// It may be called to prevent the socket being closed due to inactivity.
func (cn *Conn) KeepAlive() error {
	req := &pb.GetSocketNameRequest{
		SocketDescriptor: &cn.desc,
	}
	res := &pb.GetSocketNameReply{}
	return internal.Call(cn.ctx, "remote_socket", "GetSocketName", req, res)
}

// Put saves src to the index. If id is empty, a new ID is allocated by the
// service and returned. If id is not empty, any existing index entry for that
// ID is replaced.
//
// The ID is a human-readable ASCII string. It must contain no whitespace
// characters and not start with "!".
//
// src must be a non-nil struct pointer or implement the FieldLoadSaver
// interface.
func (x *Index) Put(c context.Context, id string, src interface{}) (string, error) {
	d, err := saveDoc(src)
	if err != nil {
		return "", err
	}
	if id != "" {
		if !validIndexNameOrDocID(id) {
			return "", fmt.Errorf("search: invalid ID %q", id)
		}
		d.Id = proto.String(id)
	}
	req := &pb.IndexDocumentRequest{
		Params: &pb.IndexDocumentParams{
			Document:  []*pb.Document{d},
			IndexSpec: &x.spec,
		},
	}
	res := &pb.IndexDocumentResponse{}
	if err := internal.Call(c, "search", "IndexDocument", req, res); err != nil {
		return "", err
	}
	if len(res.Status) > 0 {
		if s := res.Status[0]; s.GetCode() != pb.SearchServiceError_OK {
			return "", fmt.Errorf("search: %s: %s", s.GetCode(), s.GetErrorDetail())
		}
	}
	if len(res.Status) != 1 || len(res.DocId) != 1 {
		return "", fmt.Errorf("search: internal error: wrong number of results (%d Statuses, %d DocIDs)",
			len(res.Status), len(res.DocId))
	}
	return res.DocId[0], nil
}

// DeleteMulti deletes multiple tasks from a named queue.
// If a given task could not be deleted, an appengine.MultiError is returned.
func DeleteMulti(c context.Context, tasks []*Task, queueName string) error {
	taskNames := make([][]byte, len(tasks))
	for i, t := range tasks {
		taskNames[i] = []byte(t.Name)
	}
	if queueName == "" {
		queueName = "default"
	}
	req := &pb.TaskQueueDeleteRequest{
		QueueName: []byte(queueName),
		TaskName:  taskNames,
	}
	res := &pb.TaskQueueDeleteResponse{}
	if err := internal.Call(c, "taskqueue", "Delete", req, res); err != nil {
		return err
	}
	if a, b := len(req.TaskName), len(res.Result); a != b {
		return fmt.Errorf("taskqueue: internal error: requested deletion of %d tasks, got %d results", a, b)
	}
	me, any := make(appengine.MultiError, len(res.Result)), false
	for i, ec := range res.Result {
		if ec != pb.TaskQueueServiceError_OK {
			me[i] = &internal.APIError{
				Service: "taskqueue",
				Code:    int32(ec),
			}
			any = true
		}
	}
	if any {
		return me
	}
	return nil
}

// DeleteServingURL deletes the serving URL for an image.
func DeleteServingURL(c context.Context, key appengine.BlobKey) error {
	req := &pb.ImagesDeleteUrlBaseRequest{
		BlobKey: (*string)(&key),
	}
	res := &pb.ImagesDeleteUrlBaseResponse{}
	return internal.Call(c, "images", "DeleteUrlBase", req, res)
}

// DeleteMulti is a batch version of Delete.
// If any keys cannot be found, an appengine.MultiError is returned.
// Each key must be at most 250 bytes in length.
func DeleteMulti(c context.Context, key []string) error {
	if len(key) == 0 {
		return nil
	}
	req := &pb.MemcacheDeleteRequest{
		Item: make([]*pb.MemcacheDeleteRequest_Item, len(key)),
	}
	for i, k := range key {
		req.Item[i] = &pb.MemcacheDeleteRequest_Item{Key: []byte(k)}
	}
	res := &pb.MemcacheDeleteResponse{}
	if err := internal.Call(c, "memcache", "Delete", req, res); err != nil {
		return err
	}
	if len(res.DeleteStatus) != len(key) {
		return ErrServerError
	}
	me, any := make(appengine.MultiError, len(key)), false
	for i, s := range res.DeleteStatus {
		switch s {
		case pb.MemcacheDeleteResponse_DELETED:
			// OK
		case pb.MemcacheDeleteResponse_NOT_FOUND:
			me[i] = ErrCacheMiss
			any = true
		default:
			me[i] = ErrServerError
			any = true
		}
	}
	if any {
		return me
	}
	return nil
}

// Send sends a message on the channel associated with clientID.
func Send(c context.Context, clientID, message string) error {
	req := &pb.SendMessageRequest{
		ApplicationKey: &clientID,
		Message:        &message,
	}
	resp := &basepb.VoidProto{}
	return remapError(internal.Call(c, service, "SendChannelMessage", req, resp))
}

// SignBytes signs bytes using a private key unique to your application.
func SignBytes(c context.Context, bytes []byte) (keyName string, signature []byte, err error) {
	req := &pb.SignForAppRequest{BytesToSign: bytes}
	res := &pb.SignForAppResponse{}

	if err := internal.Call(c, "app_identity_service", "SignForApp", req, res); err != nil {
		return "", nil, err
	}
	return res.GetKeyName(), res.GetSignatureBytes(), nil
}

// DefaultVersion returns the default version of the specified module.
// If module is the empty string, it means the default module.
func DefaultVersion(c context.Context, module string) (string, error) {
	req := &pb.GetDefaultVersionRequest{}
	if module != "" {
		req.Module = &module
	}
	res := &pb.GetDefaultVersionResponse{}
	err := internal.Call(c, "modules", "GetDefaultVersion", req, res)
	return res.GetVersion(), err
}

// Create creates a channel and returns a token for use by the client.
// The clientID is an application-provided string used to identify the client.
func Create(c context.Context, clientID string) (token string, err error) {
	req := &pb.CreateChannelRequest{
		ApplicationKey: &clientID,
	}
	resp := &pb.CreateChannelResponse{}
	err = internal.Call(c, service, "CreateChannel", req, resp)
	token = resp.GetToken()
	return token, remapError(err)
}

// RunInBackground runs f in a background goroutine in this process.
// f is provided a context that may outlast the context provided to RunInBackground.
// This is only valid to invoke from a manually scaled module.
func RunInBackground(c context.Context, f func(c context.Context)) error {
	req := &pb.StartBackgroundRequestRequest{}
	res := &pb.StartBackgroundRequestResponse{}
	if err := internal.Call(c, "system", "StartBackgroundRequest", req, res); err != nil {
		return err
	}
	sendc <- send{res.GetRequestId(), f}
	return nil
}

// LogoutURL returns a URL that, when visited, signs the user out,
// then redirects the user to the URL specified by dest.
func LogoutURL(c context.Context, dest string) (string, error) {
	req := &pb.CreateLogoutURLRequest{
		DestinationUrl: proto.String(dest),
	}
	res := &pb.CreateLogoutURLResponse{}
	if err := internal.Call(c, "user", "CreateLogoutURL", req, res); err != nil {
		return "", err
	}
	return *res.LogoutUrl, nil
}