// Copyright (C) 2024 Quickwit, Inc.
//
// Quickwit is offered under the AGPL v3.0 and as commercial software.
// For commercial licensing, contact us at hello@quickwit.io.
//
// AGPL:
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as
// published by the Free Software Foundation, either version 3 of the
// License, or (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.

syntax = "proto3";

package quickwit.ingest.ingester;

import "quickwit/common.proto";
import "quickwit/ingest.proto";

service IngesterService {
  // Persists batches of documents to primary shards hosted on a leader.
  rpc Persist(PersistRequest) returns (PersistResponse);

  // Opens a replication stream from a leader to a follower.
  rpc OpenReplicationStream(stream SynReplicationMessage) returns (stream AckReplicationMessage);

  // Streams records from a leader or a follower. The client can optionally specify a range of positions to fetch,
  // otherwise the stream will go indefinitely or until the shard is closed.
  rpc OpenFetchStream(OpenFetchStreamRequest) returns (stream FetchMessage);

  // Streams status updates, called "observations", from an ingester.
  rpc OpenObservationStream(OpenObservationStreamRequest) returns (stream ObservationMessage);

  // Creates and initializes a set of newly opened shards. This RPC is called by the control plane on leaders.
  rpc InitShards(InitShardsRequest) returns (InitShardsResponse);

  // Only retain the shards that are listed in the request.
  // Other shards are deleted.
  rpc RetainShards(RetainShardsRequest) returns (RetainShardsResponse);

  // Truncates a set of shards at the given positions. This RPC is called by indexers on leaders AND followers.
  rpc TruncateShards(TruncateShardsRequest) returns (TruncateShardsResponse);

  // Closes a set of shards. This RPC is called by the control plane.
  rpc CloseShards(CloseShardsRequest) returns (CloseShardsResponse);

  // Decommissions the ingester.
  rpc Decommission(DecommissionRequest) returns (DecommissionResponse);
}

message RetainShardsForSource {
  quickwit.common.IndexUid index_uid = 1;
  string source_id = 2;
  repeated quickwit.ingest.ShardId shard_ids = 3;
}

message RetainShardsRequest {
  repeated RetainShardsForSource retain_shards_for_sources = 1;
}

message RetainShardsResponse {
}

message PersistRequest {
  string leader_id = 1;
  quickwit.ingest.CommitTypeV2 commit_type = 3;
  repeated PersistSubrequest subrequests = 4;
}

message PersistSubrequest {
  uint32 subrequest_id = 1;
  quickwit.common.IndexUid index_uid = 2;
  string source_id = 3;
  quickwit.ingest.ShardId shard_id = 4;
  quickwit.ingest.DocBatchV2 doc_batch = 5;
}

message PersistResponse {
  string leader_id = 1;
  repeated PersistSuccess successes = 2;
  repeated PersistFailure failures = 3;
}

message PersistSuccess {
  uint32 subrequest_id = 1;
  quickwit.common.IndexUid index_uid = 2;
  string source_id = 3;
  quickwit.ingest.ShardId shard_id = 4;
  quickwit.ingest.Position replication_position_inclusive = 5;
  uint32 num_persisted_docs = 6;
  repeated quickwit.ingest.ParseFailure parse_failures = 7;
}


enum PersistFailureReason {
  PERSIST_FAILURE_REASON_UNSPECIFIED = 0;
  PERSIST_FAILURE_REASON_SHARD_NOT_FOUND = 1;
  PERSIST_FAILURE_REASON_SHARD_CLOSED = 2;
  PERSIST_FAILURE_REASON_SHARD_RATE_LIMITED = 3;
  PERSIST_FAILURE_REASON_WAL_FULL = 4;
  PERSIST_FAILURE_REASON_TIMEOUT = 5;
}

message PersistFailure {
  uint32 subrequest_id = 1;
  quickwit.common.IndexUid index_uid = 2;
  string source_id = 3;
  quickwit.ingest.ShardId shard_id = 4;
  PersistFailureReason reason = 5;
}

message SynReplicationMessage {
  oneof message {
    OpenReplicationStreamRequest open_request = 1;
    InitReplicaRequest init_request = 2;
    ReplicateRequest replicate_request = 3;
  }
}

message AckReplicationMessage {
  oneof message {
    OpenReplicationStreamResponse open_response = 1;
    InitReplicaResponse init_response = 2;
    ReplicateResponse replicate_response = 3;
  }
}

message OpenReplicationStreamRequest {
  string leader_id = 1;
  string follower_id = 2;
  // Position of the request in the replication stream.
  uint64 replication_seqno = 3;
}

message OpenReplicationStreamResponse {
  // Position of the response in the replication stream. It should match the position of the request.
  uint64 replication_seqno = 1;
}

message InitReplicaRequest {
  Shard replica_shard = 1;
  uint64 replication_seqno = 2;
}

message InitReplicaResponse {
  uint64 replication_seqno = 1;
}

message ReplicateRequest {
  string leader_id = 1;
  string follower_id = 2;
  quickwit.ingest.CommitTypeV2 commit_type = 3;
  repeated ReplicateSubrequest subrequests = 4;
  // Position of the request in the replication stream.
  uint64 replication_seqno = 5;
}

message ReplicateSubrequest {
  uint32 subrequest_id = 1;
  quickwit.common.IndexUid index_uid = 2;
  string source_id = 3;
  quickwit.ingest.ShardId shard_id = 4;
  quickwit.ingest.Position from_position_exclusive = 5;
  ingest.DocBatchV2 doc_batch = 6;
}

message ReplicateResponse {
  string follower_id = 1;
  repeated ReplicateSuccess successes = 2;
  repeated ReplicateFailure failures = 3;
  // Position of the response in the replication stream. It should match the position of the request.
  uint64 replication_seqno = 4;
}

message ReplicateSuccess {
  uint32 subrequest_id = 1;
  quickwit.common.IndexUid index_uid = 2;
  string source_id = 3;
  quickwit.ingest.ShardId shard_id = 4;
  quickwit.ingest.Position replication_position_inclusive = 5;
}

enum ReplicateFailureReason {
  REPLICATE_FAILURE_REASON_UNSPECIFIED = 0;
  REPLICATE_FAILURE_REASON_SHARD_NOT_FOUND = 1;
  REPLICATE_FAILURE_REASON_SHARD_CLOSED = 2;
  reserved 3; // REPLICATE_FAILURE_REASON_RATE_LIMITED = 3;
  REPLICATE_FAILURE_REASON_WAL_FULL = 4;
}

message ReplicateFailure {
  uint32 subrequest_id = 1;
  quickwit.common.IndexUid index_uid = 2;
  string source_id = 3;
  quickwit.ingest.ShardId shard_id = 4;
  ReplicateFailureReason reason = 5;
}

message TruncateShardsRequest {
  string ingester_id = 1;
  repeated TruncateShardsSubrequest subrequests = 2;
}

message TruncateShardsSubrequest {
  quickwit.common.IndexUid index_uid = 1;
  string source_id = 2;
  quickwit.ingest.ShardId shard_id = 3;
  // The position up to which the shard should be truncated (inclusive).
  quickwit.ingest.Position truncate_up_to_position_inclusive = 4;
}

message TruncateShardsResponse {
  // TODO
}

message OpenFetchStreamRequest {
  string client_id = 1;
  quickwit.common.IndexUid index_uid = 2;
  string source_id = 3;
  quickwit.ingest.ShardId shard_id = 4;
  quickwit.ingest.Position from_position_exclusive = 5;
}

message FetchMessage {
  oneof message {
    FetchPayload payload = 1;
    FetchEof eof = 2;
  }
}

message FetchPayload {
  quickwit.common.IndexUid index_uid = 1;
  string source_id = 2;
  quickwit.ingest.ShardId shard_id = 3;
  quickwit.ingest.MRecordBatch mrecord_batch = 4;
  quickwit.ingest.Position from_position_exclusive = 5;
  quickwit.ingest.Position to_position_inclusive = 6;
}

message FetchEof {
  quickwit.common.IndexUid index_uid = 1;
  string source_id = 2;
  quickwit.ingest.ShardId shard_id = 3;
  quickwit.ingest.Position eof_position = 4;
}

message InitShardsRequest {
  reserved 1;
  repeated InitShardSubrequest subrequests = 2;
}

message InitShardSubrequest {
  uint32 subrequest_id = 1;
  quickwit.ingest.Shard shard = 2;
  string doc_mapping_json = 3;
  bool validate_docs = 4;
}

message InitShardsResponse {
  repeated InitShardSuccess successes = 1;
  repeated InitShardFailure failures = 2;
}

message InitShardSuccess {
  uint32 subrequest_id = 1;
  quickwit.ingest.Shard shard = 2;
}

message InitShardFailure {
  uint32 subrequest_id = 1;
  quickwit.common.IndexUid index_uid = 2;
  string source_id = 3;
  quickwit.ingest.ShardId shard_id = 4;
  // InitShardFailureReason reason = 5;
}

message CloseShardsRequest {
  reserved 1;
  repeated quickwit.ingest.ShardPKey shard_pkeys = 2;
}

message CloseShardsResponse {
  repeated quickwit.ingest.ShardPKey successes = 1;
}

message DecommissionRequest {
}

message DecommissionResponse {
}

message OpenObservationStreamRequest {
}

enum IngesterStatus {
  INGESTER_STATUS_UNSPECIFIED = 0;
  // The ingester is live but not ready yet to accept requests.
  INGESTER_STATUS_INITIALIZING = 1;
  // The ingester is ready and accepts read and write requests.
  INGESTER_STATUS_READY = 2;
  // The ingester is being decommissioned. It accepts read requests but rejects write requests
  // (open shards, persist, and replicate requests). It will transition to `Decommissioned` once
  // all shards are fully indexed.
  INGESTER_STATUS_DECOMMISSIONING = 3;
  // The ingester no longer accepts read and write requests. It does not hold any data and can
  // be safely removed from the cluster.
  INGESTER_STATUS_DECOMMISSIONED = 4;
  // The ingester failed to initialize and is not ready to accept requests.
  INGESTER_STATUS_FAILED = 5;
}

message ObservationMessage {
  string node_id = 1;
  IngesterStatus Status = 2;
}
