// 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.metastore;

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

enum SourceType {
  SOURCE_TYPE_UNSPECIFIED = 0;
  SOURCE_TYPE_CLI = 1;
  SOURCE_TYPE_FILE = 2;
  SOURCE_TYPE_INGEST_V1 = 4;
  SOURCE_TYPE_INGEST_V2 = 5;
  // Apache Kafka
  SOURCE_TYPE_KAFKA = 6;
  // Amazon Kinesis
  SOURCE_TYPE_KINESIS = 7;
  SOURCE_TYPE_NATS = 8;
  // Google Cloud Pub/Sub
  SOURCE_TYPE_PUB_SUB = 3;
  // Apache Pulsar
  SOURCE_TYPE_PULSAR = 9;
  SOURCE_TYPE_VEC = 10;
  SOURCE_TYPE_VOID = 11;
  SOURCE_TYPE_STDIN = 13;
}

// Metastore meant to manage Quickwit's indexes, their splits and delete tasks.
//
// I. Index and splits management.
//
// Quickwit needs a way to ensure that we can cleanup unused files,
// and this process needs to be resilient to any fail-stop failures.
// We rely on atomically transitioning the status of splits.
//
// The split state goes through the following life cycle:
// 1. `Staged`
//   - Start uploading the split files.
// 2. `Published`
//   - Uploading the split files is complete and the split is searchable.
// 3. `MarkedForDeletion`
//   - Mark the split for deletion.
//
// If a split has a file in the storage, it MUST be registered in the metastore,
// and its state can be as follows:
// - `Staged`: The split is almost ready. Some of its files may have been uploaded in the storage.
// - `Published`: The split is ready and published.
// - `MarkedForDeletion`: The split is marked for deletion.
//
// Before creating any file, we need to stage the split. If there is a failure, upon recovery, we
// schedule for deletion all the staged splits. A client may not necessarily remove files from
// storage right after marking it for deletion. A CLI client may delete files right away, but a
// more serious deployment should probably only delete those files after a grace period so that the
// running search queries can complete.
//
// II. Delete tasks management.
//
// A delete task is defined on a given index and by a search query. It can be
// applied to all the splits of the index.
//
// Quickwit needs a way to track that a delete task has been applied to a split. This is ensured
// by two mechanisms:
// - On creation of a delete task, we give to the task a monotically increasing opstamp (uniqueness
//   and monotonically increasing must be true at the index level).
// - When a delete task is executed on a split, that is when the documents matched by the search
//   query are removed from the splits, we update the split's `delete_opstamp` to the value of the
//   task's opstamp. This marks the split as "up-to-date" regarding this delete task. If new delete
//   tasks are added, we will know that we need to run these delete tasks on the splits as its
//   `delete_optstamp` will be inferior to the `opstamp` of the new tasks.
//
// For splits created after a given delete task, Quickwit's indexing ensures that these splits
// are created with a `delete_opstamp` equal the latest opstamp of the tasks of the
// corresponding index.
service MetastoreService {
  // Creates an index.
  //
  // This API creates a new index in the metastore.
  // An error will occur if an index that already exists in the storage is specified.
  rpc CreateIndex(CreateIndexRequest) returns (CreateIndexResponse);

  // Update an index.
  rpc UpdateIndex(UpdateIndexRequest) returns (IndexMetadataResponse);

  // Returns the `IndexMetadata` of an index identified by its IndexID or its IndexUID.
  rpc IndexMetadata(IndexMetadataRequest) returns (IndexMetadataResponse);

  // Fetches the metadata of a list of indexes identified by their Index IDs or UIDs.
  rpc IndexesMetadata(IndexesMetadataRequest) returns (IndexesMetadataResponse);

  // Gets an indexes metadatas.
  rpc ListIndexesMetadata(ListIndexesMetadataRequest) returns (ListIndexesMetadataResponse);

  // Deletes an index
  rpc DeleteIndex(DeleteIndexRequest) returns (EmptyResponse);

  // Streams splits from index.
  rpc ListSplits(ListSplitsRequest) returns (stream ListSplitsResponse);

  // Stages several splits.
  rpc StageSplits(StageSplitsRequest) returns (EmptyResponse);

  // Publishes split.
  rpc PublishSplits(PublishSplitsRequest) returns (EmptyResponse);

  // Marks splits for deletion.
  rpc MarkSplitsForDeletion(MarkSplitsForDeletionRequest) returns (EmptyResponse);

  // Deletes splits.
  rpc DeleteSplits(DeleteSplitsRequest) returns (EmptyResponse);

  // Adds source.
  rpc AddSource(AddSourceRequest) returns (EmptyResponse);

  // Toggles source.
  rpc ToggleSource(ToggleSourceRequest) returns (EmptyResponse);

  // Removes source.
  rpc DeleteSource(DeleteSourceRequest) returns (EmptyResponse);

  // Resets source checkpoint.
  rpc ResetSourceCheckpoint(ResetSourceCheckpointRequest) returns (EmptyResponse);

  // Gets last opstamp for a given `index_id`.
  rpc LastDeleteOpstamp(LastDeleteOpstampRequest) returns (LastDeleteOpstampResponse);

  // Creates a delete task.
  rpc CreateDeleteTask(DeleteQuery) returns (DeleteTask);

  // Updates splits `delete_opstamp`.
  rpc UpdateSplitsDeleteOpstamp(UpdateSplitsDeleteOpstampRequest) returns (UpdateSplitsDeleteOpstampResponse);

  // Lists delete tasks with `delete_task.opstamp` > `opstamp_start` for a given `index_id`.
  rpc ListDeleteTasks(ListDeleteTasksRequest) returns (ListDeleteTasksResponse);

  // Lists splits with `split.delete_opstamp` < `delete_opstamp` for a given `index_id`.
  rpc ListStaleSplits(ListStaleSplitsRequest) returns (ListSplitsResponse);

  // Shard API
  //
  // Note that for the file-backed metastore implementation, the requests are not processed atomically.
  // Indeed, each request comprises one or more subrequests that target different indexes and sources processed
  // independently. Responses list the requests that succeeded or failed in the fields `successes` and
  // `failures`.
  rpc OpenShards(OpenShardsRequest) returns (OpenShardsResponse);

  // Acquires a set of shards for indexing. This RPC locks the shards for publishing thanks to a publish token and only
  // the last indexer that has acquired the shards is allowed to publish. The response returns for each subrequest the
  // list of acquired shards along with the positions to index from.
  //
  // If a requested shard is missing, this method does not return an error. It should simply return the list of
  // shards that were actually acquired.
  //
  // For this reason, AcquireShards.acquire_shards may return less subresponse than there was in the request.
  // Also they may be returned in any order.
  rpc AcquireShards(AcquireShardsRequest) returns (AcquireShardsResponse);

  // Deletes a set of shards. This RPC deletes the shards from the metastore.
  // If the shard did not exist to begin with, the operation is successful and does not return any error.
  rpc DeleteShards(DeleteShardsRequest) returns (DeleteShardsResponse);

  // Deletes outdated shards. This RPC deletes the shards from the metastore.
  rpc PruneShards(PruneShardsRequest) returns (EmptyResponse);

  rpc ListShards(ListShardsRequest) returns (ListShardsResponse);

  // Index Template API
  //
  // Index templates are used to create indexes with a predefined configuration.

  // Creates an index template.
  rpc CreateIndexTemplate(CreateIndexTemplateRequest) returns (EmptyResponse);

  // Fetches an index template.
  rpc GetIndexTemplate(GetIndexTemplateRequest) returns (GetIndexTemplateResponse);

  // Finds matching index templates.
  rpc FindIndexTemplateMatches(FindIndexTemplateMatchesRequest) returns (FindIndexTemplateMatchesResponse);

  // Returns all the index templates.
  rpc ListIndexTemplates(ListIndexTemplatesRequest) returns (ListIndexTemplatesResponse);

  // Deletes index templates.
  rpc DeleteIndexTemplates(DeleteIndexTemplatesRequest) returns (EmptyResponse);
}

message EmptyResponse {
}

message CreateIndexRequest {
  string index_config_json = 2;
  repeated string source_configs_json = 3;
}

message CreateIndexResponse {
  quickwit.common.IndexUid index_uid = 1;
  string index_metadata_json = 2;
}

message UpdateIndexRequest {
  quickwit.common.IndexUid index_uid = 1;
  string search_settings_json = 2;
  optional string retention_policy_json = 3;
  string indexing_settings_json = 4;
  string doc_mapping_json = 5;
}

message ListIndexesMetadataRequest {
  reserved  1;
  // List of patterns an index should match or not match to get considered
  // An index must match at least one positive pattern (a pattern not starting
  // with a '-'), and no negative pattern (a pattern starting with a '-').
  repeated string index_id_patterns = 2;
}

message ListIndexesMetadataResponse {
  // Deprecated (v0.9.0), use `indexes_metadata_json_zstd` instead.
  optional string indexes_metadata_json_opt = 1;
  // A JSON serialized then ZSTD compressed list of `IndexMetadata`: `Vec<IndexMetadata> | JSON | ZSTD`.
  // We don't use `repeated` here to increase the compression rate and ratio.
  bytes indexes_metadata_json_zstd = 2;
}

message DeleteIndexRequest {
  quickwit.common.IndexUid index_uid = 1;
}

// Request the metadata of an index.
// Either `index_uid` or `index_id` must be specified.
//
// If both are supplied, `index_uid` is used.
message IndexMetadataRequest {
  optional string index_id = 1;
  optional quickwit.common.IndexUid index_uid = 2;
}

message IndexMetadataResponse {
  string index_metadata_serialized_json = 1;
}

message IndexesMetadataRequest {
  repeated IndexMetadataSubrequest subrequests = 1;
}

message IndexMetadataSubrequest {
  optional string index_id = 1;
  optional quickwit.common.IndexUid index_uid = 2;
}

message IndexesMetadataResponse {
  // A JSON serialized then ZSTD compressed list of `IndexMetadata`: `Vec<IndexMetadata> | JSON | ZSTD`.
  // We don't use `repeated` here to increase the compression rate and ratio.
  bytes indexes_metadata_json_zstd = 1;
  repeated IndexMetadataFailure failures = 2;
}

message IndexMetadataFailure {
  optional string index_id = 1;
  optional quickwit.common.IndexUid index_uid = 2;
  IndexMetadataFailureReason reason = 3;
}

enum IndexMetadataFailureReason {
  INDEX_METADATA_FAILURE_REASON_UNSPECIFIED = 0;
  INDEX_METADATA_FAILURE_REASON_NOT_FOUND = 1;
  INDEX_METADATA_FAILURE_REASON_INTERNAL = 2;
}

message ListSplitsRequest {
  // Predicate used to filter splits.
  // The predicate is expressed as a JSON serialized
  // `ListSplitsQuery`.
  string query_json = 1;
}

message ListSplitsResponse {
  // TODO use repeated and encode splits json individually.
  string splits_serialized_json = 1;
}

message StageSplitsRequest {
  quickwit.common.IndexUid index_uid = 1;
  string split_metadata_list_serialized_json = 2;
}

message PublishSplitsRequest {
  quickwit.common.IndexUid index_uid = 1;
  repeated string staged_split_ids = 2;
  repeated string replaced_split_ids = 3;
  optional string index_checkpoint_delta_json_opt = 4;
  optional string publish_token_opt = 5;
}

message MarkSplitsForDeletionRequest {
  quickwit.common.IndexUid index_uid = 2;
  repeated string split_ids = 3;
}

message DeleteSplitsRequest {
  quickwit.common.IndexUid index_uid = 2;
  repeated string split_ids = 3;
}

message AddSourceRequest {
  quickwit.common.IndexUid index_uid = 1;
  string source_config_json = 2;
}

message ToggleSourceRequest {
  quickwit.common.IndexUid index_uid = 1;
  string source_id = 2;
  bool enable = 3;
}

message DeleteSourceRequest {
  quickwit.common.IndexUid index_uid = 1;
  string source_id = 2;
}

message ResetSourceCheckpointRequest {
  quickwit.common.IndexUid index_uid = 1;
  string source_id = 2;
}

//
// Delete tasks API.
//

message DeleteTask {
  int64 create_timestamp = 1;
  uint64 opstamp = 2;
  DeleteQuery delete_query = 3;
}

message DeleteQuery {
  reserved 4, 5;

  // Index UID.
  quickwit.common.IndexUid index_uid = 1;
  // If set, restrict search to documents with a `timestamp >= start_timestamp`.
  optional int64 start_timestamp = 2;
  // If set, restrict search to documents with a `timestamp < end_timestamp``.
  optional int64 end_timestamp = 3;
  // Query AST serialized in JSON
  string query_ast = 6;
}

message UpdateSplitsDeleteOpstampRequest {
  quickwit.common.IndexUid index_uid = 1;
  repeated string split_ids = 2;
  uint64 delete_opstamp = 3;
}

message UpdateSplitsDeleteOpstampResponse {}

message LastDeleteOpstampRequest {
  quickwit.common.IndexUid index_uid = 1;
}

message LastDeleteOpstampResponse {
  uint64 last_delete_opstamp = 1;
}

message ListStaleSplitsRequest {
  quickwit.common.IndexUid index_uid = 1;
  uint64 delete_opstamp = 2;
  uint64 num_splits = 3;
}

message ListDeleteTasksRequest {
  quickwit.common.IndexUid index_uid = 1;
  uint64 opstamp_start = 2;
}

message ListDeleteTasksResponse {
  repeated DeleteTask delete_tasks = 1;
}

//
// Shard API
//

message OpenShardsRequest {
  repeated OpenShardSubrequest subrequests = 1;
}

message OpenShardSubrequest {
  uint32 subrequest_id = 1;
  quickwit.common.IndexUid index_uid = 2;
  string source_id = 3;
  quickwit.ingest.ShardId shard_id = 4;
  string leader_id = 5;
  optional string follower_id = 6;
  quickwit.common.DocMappingUid doc_mapping_uid = 7;
  optional string publish_token = 8;
}

message OpenShardsResponse {
  repeated OpenShardSubresponse subresponses = 1;
}

message OpenShardSubresponse {
  reserved 2, 3;

  uint32 subrequest_id = 1;
  quickwit.ingest.Shard open_shard = 4;
}

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

message AcquireShardsResponse {
  // List of acquired shards, in no specific order.
  repeated quickwit.ingest.Shard acquired_shards = 3;
}

message DeleteShardsRequest {
  quickwit.common.IndexUid index_uid = 1;
  string source_id = 2;
  repeated quickwit.ingest.ShardId shard_ids = 3;
  // If false, only shards at EOF positions will be deleted.
  bool force = 4;
}

message DeleteShardsResponse {
  quickwit.common.IndexUid index_uid = 1;
  string source_id = 2;
  // List of shard IDs that were successfully deleted.
  repeated quickwit.ingest.ShardId successes = 3;
  // List of shard IDs that could not be deleted because `force` was set to `false` in the request,
  // and the shards are not at EOF, i.e., not fully indexed.
  repeated quickwit.ingest.ShardId failures = 4;
}

message PruneShardsRequest {
  quickwit.common.IndexUid index_uid = 1;
  string source_id = 2;
  // The maximum age of the shards to keep, in seconds.
  optional uint32 max_age_secs = 5;
  // The maximum number of the shards to keep. Delete older shards first.
  optional uint32 max_count = 6;
  // The interval between two pruning operations, in seconds.
  optional uint32 interval_secs = 7;
}

message ListShardsRequest {
  repeated ListShardsSubrequest subrequests = 1;
}

message ListShardsSubrequest {
  quickwit.common.IndexUid index_uid = 1;
  string source_id = 2;
  optional quickwit.ingest.ShardState shard_state = 3;
}

message ListShardsResponse {
  repeated ListShardsSubresponse subresponses = 1;
}

message ListShardsSubresponse {
  quickwit.common.IndexUid index_uid = 1;
  string source_id = 2;
  repeated quickwit.ingest.Shard shards = 3;
}

//
// Index Template API
//

message CreateIndexTemplateRequest {
  string index_template_json = 1;
  bool overwrite = 2;
}

message GetIndexTemplateRequest {
  string template_id = 1;
}

message GetIndexTemplateResponse {
  string index_template_json = 1;
}

message FindIndexTemplateMatchesRequest {
  repeated string index_ids = 1;
}

message FindIndexTemplateMatchesResponse {
  repeated IndexTemplateMatch matches = 1;
}

message IndexTemplateMatch {
  string index_id = 1;
  string template_id = 2;
  string index_template_json = 3;
}

message ListIndexTemplatesRequest {
}

message ListIndexTemplatesResponse {
  repeated string index_templates_json = 1;
}

message DeleteIndexTemplatesRequest {
  repeated string template_ids = 1;
}
