// Licensed to the Apache Software Foundation (ASF) under one or more
// contributor license agreements.  See the NOTICE file distributed with
// this work for additional information regarding copyright ownership.
// The ASF licenses this file to You under the Apache License, Version 2.0
// (the "License"); you may not use this file except in compliance with
// the License.  You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

syntax = "proto3";

import "google/protobuf/duration.proto";
import "google/protobuf/timestamp.proto";

import "apache/rocketmq/v2/definition.proto";

package apache.rocketmq.v2;

option csharp_namespace = "Apache.Rocketmq.V2";
option java_multiple_files = true;
option java_package = "apache.rocketmq.v2";
option java_generate_equals_and_hash = true;
option java_string_check_utf8 = true;
option java_outer_classname = "MQService";

// Topics are destination of messages to publish to or subscribe from. Similar
// to domain names, they will be addressable after resolution through the
// provided access point.
//
// Access points are usually the addresses of name servers, which fulfill
// service discovery, load-balancing and other auxiliary services. Name servers
// receive periodic heartbeats from affiliate brokers and erase those which
// failed to maintain alive status.
//
// Name servers answer queries of QueryRouteRequest, responding clients with
// addressable message-queues, which they may directly publish messages to or
// subscribe messages from.
//
// QueryRouteRequest shall include source endpoints, aka, configured
// access-point, which annotates tenant-id, instance-id or other
// vendor-specific settings. Purpose-built name servers may respond customized
// results based on these particular requirements.
message QueryRouteRequest {
  Resource topic = 1;

  Endpoints endpoints = 2;
}

message QueryRouteResponse {
  Status status = 1;

  repeated MessageQueue message_queues = 2;
}

message SendMessageRequest {
  repeated Message messages = 1;
  MessageQueue message_queue = 2;
}

message SendMessageResponse {
  Status status = 1;
  repeated SendReceipt receipts = 2;
}

message QueryAssignmentRequest {
  Resource topic = 1;
  Resource group = 2;
  string client_id = 3;

  // Service access point
  Endpoints endpoints = 4;
}

message QueryAssignmentResponse {
  Status status = 1;
  repeated Assignment assignments = 2;
}

message ReceiveMessageRequest {
  Resource group = 1;
  string client_id = 2;
  MessageQueue message_queue = 3;
  FilterExpression filter_expression = 4;
  google.protobuf.Timestamp initialization_timestamp = 5;
  int32 batch_size = 6;
  google.protobuf.Duration invisible_duration = 7;
  google.protobuf.Duration await_duration = 8;
  bool fifo = 9;
}

message ReceiveMessageResponse {
  Status status = 1;
  repeated Message messages = 2;
  google.protobuf.Timestamp delivery_timestamp = 3;
  google.protobuf.Duration invisible_duration = 4;
}

message AckMessageRequest {
  Resource group = 1;
  Resource topic = 2;
  string client_id = 3;
  string receipt_handle = 4;
  string message_id = 5;
}

message AckMessageResponse { Status status = 1; }

message ForwardMessageToDeadLetterQueueRequest {
  Resource group = 1;
  Resource topic = 2;
  string client_id = 3;
  string receipt_handle = 4;
  string message_id = 5;
  int32 delivery_attempt = 6;
  int32 max_delivery_attempts = 7;
}

message ForwardMessageToDeadLetterQueueResponse { Status status = 1; }

message HeartbeatRequest {
  string client_id = 1;
  Resource group = 2;
}

message HeartbeatResponse { Status status = 1; }

message EndTransactionRequest {
  Resource group = 1;
  string message_id = 2;
  string transaction_id = 3;
  TransactionResolution resolution = 4;
  TransactionSource source = 5;
  string trace_context = 6;
}

message EndTransactionResponse { Status status = 1; }

message QueryOffsetRequest {
  MessageQueue message_queue = 1;
  QueryOffsetPolicy policy = 2;
  google.protobuf.Timestamp time_point = 3;
}

message QueryOffsetResponse {
  Status status = 1;
  int64 offset = 2;
}

message PullMessageRequest {
  Resource group = 1;
  MessageQueue message_queue = 2;
  int64 offset = 3;
  int32 batch_size = 4;
  google.protobuf.Duration await_time = 5;
  FilterExpression filter_expression = 6;
  string client_id = 7;
}

message PullMessageResponse {
  Status status = 1;
  int64 min_offset = 2;
  int64 next_offset = 3;
  int64 max_offset = 4;
  repeated Message messages = 5;
}

message PrintThreadStackTraceCommand { int64 command_id = 1; }

message ThreadStackTrace {
  int64 command_id = 1;
  string thread_stack_trace = 2;
}

message VerifyMessageCommand {
  int64 command_id = 1;
  Message message = 2;
}

message VerifyMessageResult {
  int64 command_id = 1;
  Status status = 2;
}

message RecoverOrphanedTransactionCommand {
  int64 command_id = 1;
  Message orphaned_transactional_message = 2;
  string transaction_id = 3;
}

message Settings {
  string client_id = 1;
  string access_point = 2;
  Publish publish = 3;
  Subscription subscription = 4;
  Authentication authentication = 5;
}

message TelemetryCommand {
  oneof command {
    Settings settings = 1;

    // Request client to recover the orphaned transaction message.
    RecoverOrphanedTransactionCommand recover_orphaned_transaction_command = 2;

    // Request client to print thread stack trace.
    PrintThreadStackTraceCommand print_thread_stack_trace_command = 3;

    ThreadStackTrace thread_stack_trace = 4;

    // Request client to verify the consumption of the appointed message.
    VerifyMessageCommand verify_message_command = 5;

    VerifyMessageResult verify_message_result = 6;
  }
}

message NotifyClientTerminationRequest {
  Resource group = 1;
  string client_id = 2;
}

message NotifyClientTerminationResponse { Status status = 1; }

message ChangeInvisibleDurationRequest {
  Resource group = 1;
  Resource topic = 2;

  // Unique receipt handle to identify message to change
  string receipt_handle = 3;

  // New invisible duration
  google.protobuf.Duration invisible_duration = 4;
}

message ChangeInvisibleDurationResponse {
  Status status = 1;

  // Server may generate a new receipt handle for the message.
  string receipt_handle = 2;
}

// For all the RPCs in MessagingService, the following error handling policies
// apply:
//
// If the request doesn't bear a valid authentication credential, return a
// response with common.status.code == `UNAUTHENTICATED`. If the authenticated
// user is not granted with sufficient permission to execute the requested
// operation, return a response with common.status.code == `PERMISSION_DENIED`.
// If the per-user-resource-based quota is exhausted, return a response with
// common.status.code == `RESOURCE_EXHAUSTED`. If any unexpected server-side
// errors raise, return a response with common.status.code == `INTERNAL`.
service MessagingService {

  // Queries the route entries of the requested topic in the perspective of the
  // given endpoints. On success, servers should return a collection of
  // addressable message-queues. Note servers may return customized route
  // entries based on endpoints provided.
  //
  // If the requested topic doesn't exist, returns `NOT_FOUND`.
  // If the specific endpoints is emtpy, returns `INVALID_ARGUMENT`.
  rpc QueryRoute(QueryRouteRequest) returns (QueryRouteResponse) {}

  // Producer or consumer sends HeartbeatRequest to servers periodically to
  // keep-alive. Additionally, it also reports client-side configuration,
  // including topic subscription, load-balancing group name, etc.
  //
  // Returns `OK` if success.
  //
  // If a client specifies a language that is not yet supported by servers,
  // returns `INVALID_ARGUMENT`
  rpc Heartbeat(HeartbeatRequest) returns (HeartbeatResponse) {}

  // Delivers messages to brokers.
  // Clients may further:
  // 1. Refine a message destination to message-queues which fulfills parts of
  // FIFO semantic;
  // 2. Flag a message as transactional, which keeps it invisible to consumers
  // until it commits;
  // 3. Time a message, making it invisible to consumers till specified
  // time-point;
  // 4. And more...
  //
  // Returns message-id or transaction-id with status `OK` on success.
  //
  // If the destination topic doesn't exist, returns `NOT_FOUND`.
  rpc SendMessage(SendMessageRequest) returns (SendMessageResponse) {}

  // Queries the assigned route info of a topic for current consumer,
  // the returned assignment result is decided by server-side load balancer.
  //
  // If the corresponding topic doesn't exist, returns `NOT_FOUND`.
  // If the specific endpoints is empty, returns `INVALID_ARGUMENT`.
  rpc QueryAssignment(QueryAssignmentRequest)
      returns (QueryAssignmentResponse) {}

  // Receives messages from the server in batch manner, returns a set of
  // messages if success. The received messages should be acked or nacked after
  // processed.
  //
  // If the pending concurrent receive requests exceed the quota of the given
  // consumer group, returns `UNAVAILABLE`. If the upstream store server hangs,
  // return `DEADLINE_EXCEEDED` in a timely manner. If the corresponding topic
  // or consumer group doesn't exist, returns `NOT_FOUND`. If there is no new
  // message in the specific topic, returns `OK` with an empty message set.
  // Please note that client may suffer from false empty responses.
  rpc ReceiveMessage(ReceiveMessageRequest) returns (ReceiveMessageResponse) {}

  // Acknowledges the message associated with the `receipt_handle` or `offset`
  // in the `AckMessageRequest`, it means the message has been successfully
  // processed. Returns `OK` if the message server remove the relevant message
  // successfully.
  //
  // If the given receipt_handle is illegal or out of date, returns
  // `INVALID_ARGUMENT`.
  rpc AckMessage(AckMessageRequest) returns (AckMessageResponse) {}

  // Forwards one message to dead letter queue if the DeadLetterPolicy is
  // triggered by this message at client-side, return `OK` if success.
  rpc ForwardMessageToDeadLetterQueue(ForwardMessageToDeadLetterQueueRequest)
      returns (ForwardMessageToDeadLetterQueueResponse) {}

  // Commits or rollback one transactional message.
  rpc EndTransaction(EndTransactionRequest) returns (EndTransactionResponse) {}

  // Queries the offset of the specific message queue, returns the offset with
  // `OK` if success. The message server should maintain a numerical offset for
  // each message in a message-queue.
  rpc QueryOffset(QueryOffsetRequest) returns (QueryOffsetResponse) {}

  // Pulls messages from the specific message-queue, returns a set of messages
  // with next pull offset. The pulled messages can't be acked or nacked, while
  // the client is responsible for manage offsets for consumer, typically update
  // consume offset to local memory or a third-party storage service.
  //
  // If the pending concurrent receive requests exceed the quota of the given
  // consumer group, returns `UNAVAILABLE`. If the upstream store server hangs,
  // return `DEADLINE_EXCEEDED` in a timely manner. If the corresponding topic
  // or consumer group doesn't exist, returns `NOT_FOUND`. If there is no new
  // message in the specific topic, returns `OK` with an empty message set.
  // Please note that client may suffer from false empty responses.
  rpc PullMessage(PullMessageRequest) returns (PullMessageResponse) {}

  // Once a client starts, it would immediately establishes bi-lateral stream
  // RPCs with brokers, reporting its settings as the initiative command.
  //
  // When servers have need of inspecting client status, they would issue
  // telemetry commands to clients. After executing recieved instructions,
  // clients shall report command execution results through client-side streams.
  rpc Telemetry(stream TelemetryCommand) returns (stream TelemetryCommand) {}

  // Notify the server that the client is terminated.
  rpc NotifyClientTermination(NotifyClientTerminationRequest)
      returns (NotifyClientTerminationResponse) {}

  // Once a message is retrieved from consume queue on behalf of the group, it
  // will be kept invisible to other clients of the same group for a period of
  // time. The message is supposed to be processed within the invisible
  // duration. If the client, which is in charge of the invisible message, is
  // not capable of processing the message timely, it may use
  // ChangeInvisibleDuration to lengthen invisible duration.
  rpc ChangeInvisibleDuration(ChangeInvisibleDurationRequest)
      returns (ChangeInvisibleDurationResponse) {}
}