syntax = "proto3";

package meta;

import "backup_service.proto";
import "catalog.proto";
import "common.proto";
import "hummock.proto";
import "secret.proto";
import "source.proto";
import "stream_plan.proto";
import "user.proto";

option java_package = "com.risingwave.proto";
option optimize_for = SPEED;

message GetTelemetryInfoRequest {}

message TelemetryInfoResponse {
  optional string tracking_id = 1;
}

service TelemetryInfoService {
  // Request telemetry info from meta node
  rpc GetTelemetryInfo(GetTelemetryInfoRequest) returns (TelemetryInfoResponse);
}

message HeartbeatRequest {
  uint32 node_id = 1;
  reserved 2;
  reserved "resource";
}

message HeartbeatResponse {
  common.Status status = 1;
}

service HeartbeatService {
  rpc Heartbeat(HeartbeatRequest) returns (HeartbeatResponse);
}

// Fragments of a Streaming Job.
// It's for all kinds of streaming jobs, and ideally should be called `StreamingJobFragments`.
// It's not the same as a storage table correlated with a `TableCatalog`.
message TableFragments {
  // The state of the fragments of this table
  enum State {
    UNSPECIFIED = 0;
    // The streaming job is initial.
    INITIAL = 1;
    // The streaming job is creating.
    CREATING = 2;
    // The streaming job has been created.
    CREATED = 3;
  }
  // Runtime information of an actor
  message ActorStatus {
    // Current on which worker
    common.ActorLocation location = 1;
    // Field 2 for actor state has been reserved for backward compatibility.
    reserved 2;
    reserved "state";
  }
  message Fragment {
    enum FragmentDistributionType {
      UNSPECIFIED = 0;
      SINGLE = 1;
      HASH = 2;
    }
    uint32 fragment_id = 1;
    // Bitwise-OR of FragmentTypeFlags
    uint32 fragment_type_mask = 2;
    FragmentDistributionType distribution_type = 3;
    repeated stream_plan.StreamActor actors = 4;

    // NOTE: vnode_mapping is deprecated, we will generate the vnode_mapping by actors' bitmaps
    // Vnode mapping (which should be set in upstream dispatcher) of the fragment.
    // This field is always set to `Some`. For singleton, the parallel unit for all vnodes will be the same.
    reserved 5;
    reserved "vnode_mapping";

    repeated uint32 state_table_ids = 6;
    // Note that this can be derived backwards from the upstream actors of the Actor held by the Fragment,
    // but in some scenarios (e.g. Scaling) it will lead to a lot of duplicate code,
    // so we pre-generate and store it here, this member will only be initialized when creating the Fragment
    // and modified when creating the mv-on-mv
    repeated uint32 upstream_fragment_ids = 7;

    // Total vnode count of the fragment (then all internal tables).
    // Duplicated from the length of the vnode bitmap in any actor of the fragment.
    //
    // Can be unset if the fragment is created in older versions where variable vnode count is not
    // supported, in which case a default value of 256 (or 1 for singleton) should be used.
    // Use `VnodeCountCompat::vnode_count` to access it.
    optional uint32 maybe_vnode_count = 8;

    stream_plan.StreamNode nodes = 9;
  }
  // The id of the streaming job.
  uint32 table_id = 1;
  State state = 2;
  map<uint32, Fragment> fragments = 3;
  map<uint32, ActorStatus> actor_status = 4;

  reserved 5;
  reserved "actor_splits";

  stream_plan.StreamContext ctx = 6;

  reserved 7;
  reserved "parallelism";
  // The max parallelism specified when the streaming job was created, i.e., expected vnode count.
  //
  // The reason for persisting this value is mainly to check if a parallelism change (via `ALTER
  // .. SET PARALLELISM`) is valid, so that the behavior can be consistent with the creation of
  // the streaming job.
  //
  // Note that the actual vnode count, denoted by `vnode_count` in `fragments`, may be different
  // from this value (see `StreamFragmentGraph.max_parallelism` for more details.). As a result,
  // checking the parallelism change with this value can be inaccurate in some cases. However,
  // when generating resizing plans, we still take the `vnode_count` of each fragment into account.
  //
  // Can be unset if the fragment is created in older versions where variable vnode count is not
  // supported, in which case a default value of 256 should be used.
  optional uint32 max_parallelism = 10;

  // Actors of a materialize view, sink, or table can only be scheduled on nodes with matching node_label.
  string node_label = 8;

  // If this is a materialized view: True if backfill is done, else false.
  // If this is a regular table: Always true.
  bool backfill_done = 9;
}

/// Worker slot mapping with fragment id, used for notification.
message FragmentWorkerSlotMapping {
  uint32 fragment_id = 1;
  common.WorkerSlotMapping mapping = 2;
}

message FragmentWorkerSlotMappings {
  repeated FragmentWorkerSlotMapping mappings = 1;
}

// TODO: remove this when dashboard refactored.
message ActorLocation {
  common.WorkerNode node = 1;
  repeated stream_plan.StreamActor actors = 2;
}

message MigrationPlan {
  // NOTE: parallel_unit_migration_plan is deprecated, using worker_slot_migration_plan instead
  // map<parallel_unit_id, parallel_unit>, the plan indicates that the actors will be migrated from old parallel unit to the new one.
  reserved 1;
  reserved "parallel_unit_migration_plan";

  // map<src_worker_slot_id, dst_worker_slot_id>, the plan indicates that the actors will be migrated from old worker_slot to the new one.
  map<uint64, uint64> worker_slot_migration_plan = 2;
}

message FlushRequest {
  uint32 database_id = 1;
}

message FlushResponse {
  common.Status status = 1;
  uint64 hummock_version_id = 2;
}

message PauseRequest {}

message PauseResponse {}

message ResumeRequest {}

message ResumeResponse {}

message RefreshRequest {
  uint32 table_id = 1;
  uint32 associated_source_id = 2;
}

message RefreshResponse {
  common.Status status = 1;
}

message CancelCreatingJobsRequest {
  message CreatingJobInfo {
    uint32 database_id = 1;
    uint32 schema_id = 2;
    string name = 3;
  }

  message CreatingJobInfos {
    repeated CreatingJobInfo infos = 1;
  }

  message CreatingJobIds {
    repeated uint32 job_ids = 1;
  }

  oneof jobs {
    CreatingJobInfos infos = 1;
    CreatingJobIds ids = 2;
  }
}

message CancelCreatingJobsResponse {
  common.Status status = 1;
  repeated uint32 canceled_jobs = 2;
}

message ListTableFragmentsRequest {
  repeated uint32 table_ids = 1;
}

message ListTableFragmentsResponse {
  message ActorInfo {
    uint32 id = 1;
    stream_plan.StreamNode node = 2;
    repeated stream_plan.Dispatcher dispatcher = 3;
  }
  message FragmentInfo {
    uint32 id = 1;
    repeated ActorInfo actors = 4;
  }
  message TableFragmentInfo {
    repeated FragmentInfo fragments = 1;
    stream_plan.StreamContext ctx = 2;
  }
  map<uint32, TableFragmentInfo> table_fragments = 1;
}

message GetFragmentByIdRequest {
  uint32 fragment_id = 1;
}

message GetFragmentByIdResponse {
  FragmentDistribution distribution = 1;
}

message GetFragmentVnodesRequest {
  uint32 fragment_id = 1;
}

message GetFragmentVnodesResponse {
  message ActorVnodes {
    uint32 actor_id = 1;
    repeated uint32 vnode_indices = 2;
  }
  repeated ActorVnodes actor_vnodes = 1;
}

message GetActorVnodesRequest {
  uint32 actor_id = 1;
}

message GetActorVnodesResponse {
  repeated uint32 vnode_indices = 1;
}

message ListStreamingJobStatesRequest {}

message ListStreamingJobStatesResponse {
  message StreamingJobState {
    uint32 table_id = 1;
    TableFragments.State state = 2;
    TableParallelism parallelism = 3;
    uint32 max_parallelism = 4;
    string name = 5;
    string resource_group = 6;
    string config_override = 9;
    uint32 database_id = 7;
    uint32 schema_id = 8;
    optional string adaptive_parallelism_strategy = 10;
  }
  repeated StreamingJobState states = 1;
}

message ListFragmentDistributionRequest {
  optional bool include_node = 1;
}

message FragmentDistribution {
  uint32 fragment_id = 1;
  uint32 table_id = 2;
  TableFragments.Fragment.FragmentDistributionType distribution_type = 3;
  repeated uint32 state_table_ids = 4;
  repeated uint32 upstream_fragment_ids = 5;
  uint32 fragment_type_mask = 6;
  uint32 parallelism = 7;
  uint32 vnode_count = 8;
  stream_plan.StreamNode node = 9;
  string parallelism_policy = 10;
}

message ListFragmentDistributionResponse {
  repeated FragmentDistribution distributions = 1;
}

// List fragments containing stream scans, and corresponding
// to a streaming job in creation.
message ListCreatingFragmentDistributionRequest {
  optional bool include_node = 1;
}

// List fragments containing stream scans, and corresponding
// to a streaming job in creation.
message ListCreatingFragmentDistributionResponse {
  repeated FragmentDistribution distributions = 1;
}

message ListSinkLogStoreTablesRequest {}

message ListSinkLogStoreTablesResponse {
  message SinkLogStoreTable {
    uint32 sink_id = 1;
    uint32 internal_table_id = 2;
  }
  repeated SinkLogStoreTable tables = 1;
}

message ListActorStatesRequest {}

message ListActorStatesResponse {
  message ActorState {
    uint32 actor_id = 1;
    uint32 fragment_id = 2;
    reserved 3;
    reserved "parallel_unit_id";
    reserved 4;
    reserved "state";
    uint32 worker_id = 5;
  }
  repeated ActorState states = 1;
}

message ListActorSplitsRequest {}

message ListActorSplitsResponse {
  enum FragmentType {
    UNSPECIFIED = 0;
    NON_SHARED_SOURCE = 1;
    SHARED_SOURCE = 2;
    SHARED_SOURCE_BACKFILL = 3;
  }
  message ActorSplit {
    uint32 actor_id = 1;
    uint32 fragment_id = 2;
    uint32 source_id = 3;
    string split_id = 4;
    FragmentType fragment_type = 5;
  }
  repeated ActorSplit actor_splits = 1;
}

message ObjectDependency {
  uint32 object_id = 1;
  uint32 referenced_object_id = 2;
  common.ObjectType referenced_object_type = 3;
}

enum ThrottleTarget {
  THROTTLE_TARGET_UNSPECIFIED = 0;
  SOURCE = 1;
  MV = 2;
  TABLE = 3;
  SINK = 4;
  FRAGMENT = 5;
}

message ApplyThrottleRequest {
  ThrottleTarget throttle_target = 1;
  common.ThrottleType throttle_type = 2;
  uint32 id = 3;
  optional uint32 rate = 4;
}

message ApplyThrottleResponse {
  common.Status status = 1;
}

message RecoverRequest {}

message RecoverResponse {}

message AlterConnectorPropsRequest {
  message AlterIcebergTableIds {
    uint32 source_id = 1;
    uint32 sink_id = 2;
  }
  enum AlterConnectorPropsObject {
    UNSPECIFIED = 0;
    SOURCE = 1;
    SINK = 2;
    CONNECTION = 3;
    ICEBERG_TABLE = 4;
  }

  uint32 object_id = 1;
  map<string, string> changed_props = 2;
  map<string, secret.SecretRef> changed_secret_refs = 3;
  optional uint32 connector_conn_ref = 4;
  AlterConnectorPropsObject object_type = 5;
  oneof extra_options {
    AlterIcebergTableIds alter_iceberg_table_ids = 6;
  }
}

message AlterConnectorPropsResponse {}

// Options for controlling the safe property update workflow
message PropertyUpdateOptions {
  // If true, reset split assignments after updating properties
  // This causes splits to be re-discovered from upstream
  bool reset_splits = 1;
}

// Request for orchestrated source property update with pause/resume
message AlterSourcePropertiesSafeRequest {
  uint32 source_id = 1;
  map<string, string> changed_props = 2;
  map<string, secret.SecretRef> changed_secret_refs = 3;
  PropertyUpdateOptions options = 4;
}

message AlterSourcePropertiesSafeResponse {}

// Request to reset source split assignments (UNSAFE - admin only)
message ResetSourceSplitsRequest {
  uint32 source_id = 1;
}

message ResetSourceSplitsResponse {}

// Request to inject specific offsets into source splits (UNSAFE - admin only)
message InjectSourceOffsetsRequest {
  uint32 source_id = 1;
  // Split ID -> offset (JSON-encoded based on connector type)
  map<string, string> split_offsets = 2;
}

message InjectSourceOffsetsResponse {
  repeated string applied_split_ids = 1;
}

message ListCdcProgressRequest {}

message ListCdcProgressResponse {
  message CdcProgress {
    uint64 split_total_count = 1;
    uint64 split_backfilled_count = 2;
    uint64 split_completed_count = 3;
  }
  map<uint32, CdcProgress> cdc_progress = 1;
}

message ListRefreshTableStatesRequest {}

message ListRefreshTableStatesResponse {
  message RefreshTableState {
    uint32 table_id = 1;
    string current_status = 2;
    optional string last_trigger_time = 3;
    optional int64 trigger_interval_secs = 4;
    optional string last_success_time = 5;
  }
  repeated RefreshTableState states = 1;
}

message ListIcebergCompactionStatusRequest {}

message ListIcebergCompactionStatusResponse {
  message IcebergCompactionStatus {
    uint32 sink_id = 1;
    string task_type = 2;
    uint64 trigger_interval_sec = 3;
    uint64 trigger_snapshot_count = 4;
    string schedule_state = 5;
    optional uint64 next_compaction_after_sec = 6;
    // V1 approximates pending snapshot backlog with in-memory sink commit backlog.
    optional uint64 pending_snapshot_count = 7;
    bool is_triggerable = 8;
  }
  repeated IcebergCompactionStatus statuses = 1;
}

service StreamManagerService {
  rpc Flush(FlushRequest) returns (FlushResponse);
  rpc Pause(PauseRequest) returns (PauseResponse);
  rpc Resume(ResumeRequest) returns (ResumeResponse);
  rpc Refresh(RefreshRequest) returns (RefreshResponse);
  rpc CancelCreatingJobs(CancelCreatingJobsRequest) returns (CancelCreatingJobsResponse);
  rpc ListTableFragments(ListTableFragmentsRequest) returns (ListTableFragmentsResponse);
  rpc ListStreamingJobStates(ListStreamingJobStatesRequest) returns (ListStreamingJobStatesResponse);
  rpc ListFragmentDistribution(ListFragmentDistributionRequest) returns (ListFragmentDistributionResponse);
  rpc ListCreatingFragmentDistribution(ListCreatingFragmentDistributionRequest) returns (ListCreatingFragmentDistributionResponse);
  rpc ListSinkLogStoreTables(ListSinkLogStoreTablesRequest) returns (ListSinkLogStoreTablesResponse);
  rpc ListActorStates(ListActorStatesRequest) returns (ListActorStatesResponse);
  rpc ListActorSplits(ListActorSplitsRequest) returns (ListActorSplitsResponse);
  rpc ApplyThrottle(ApplyThrottleRequest) returns (ApplyThrottleResponse);
  rpc Recover(RecoverRequest) returns (RecoverResponse);
  rpc ListRateLimits(ListRateLimitsRequest) returns (ListRateLimitsResponse);
  rpc AlterConnectorProps(AlterConnectorPropsRequest) returns (AlterConnectorPropsResponse);
  rpc AlterSourcePropertiesSafe(AlterSourcePropertiesSafeRequest) returns (AlterSourcePropertiesSafeResponse);
  rpc ResetSourceSplits(ResetSourceSplitsRequest) returns (ResetSourceSplitsResponse);
  rpc InjectSourceOffsets(InjectSourceOffsetsRequest) returns (InjectSourceOffsetsResponse);
  rpc GetFragmentById(GetFragmentByIdRequest) returns (GetFragmentByIdResponse);
  rpc GetFragmentVnodes(GetFragmentVnodesRequest) returns (GetFragmentVnodesResponse);
  rpc GetActorVnodes(GetActorVnodesRequest) returns (GetActorVnodesResponse);
  rpc SetSyncLogStoreAligned(SetSyncLogStoreAlignedRequest) returns (SetSyncLogStoreAlignedResponse);
  rpc ListCdcProgress(ListCdcProgressRequest) returns (ListCdcProgressResponse);
  rpc ListUnmigratedTables(ListUnmigratedTablesRequest) returns (ListUnmigratedTablesResponse);
  rpc ListRefreshTableStates(ListRefreshTableStatesRequest) returns (ListRefreshTableStatesResponse);
  rpc ListIcebergCompactionStatus(ListIcebergCompactionStatusRequest) returns (ListIcebergCompactionStatusResponse);
}

// Below for cluster service.
message AddWorkerNodeRequest {
  common.WorkerType worker_type = 1;
  common.HostAddress host = 2;
  reserved 3;
  reserved 4;
  common.WorkerNode.Resource resource = 5;
  common.WorkerNode.Property property = 6;
}

message AddWorkerNodeResponse {
  optional uint32 node_id = 2;
  string cluster_id = 4;
}

message ActivateWorkerNodeRequest {
  common.HostAddress host = 1;
  uint32 node_id = 2;
}

message ActivateWorkerNodeResponse {
  common.Status status = 1;
}

message DeleteWorkerNodeRequest {
  common.HostAddress host = 1;
}

message DeleteWorkerNodeResponse {
  common.Status status = 1;
}

message ListAllNodesRequest {
  optional common.WorkerType worker_type = 1;
  // Whether to include nodes still starting
  bool include_starting_nodes = 2;
}

message ListAllNodesResponse {
  common.Status status = 1;
  repeated common.WorkerNode nodes = 2;
}

message GetClusterRecoveryStatusRequest {}

enum RecoveryStatus {
  STATUS_UNSPECIFIED = 0;
  STATUS_STARTING = 1;
  STATUS_RECOVERING = 2;
  STATUS_RUNNING = 3;
}

message GetClusterRecoveryStatusResponse {
  RecoveryStatus status = 1;
}

message GetMetaStoreInfoRequest {}

message GetMetaStoreInfoResponse {
  string meta_store_endpoint = 1;
}

service ClusterService {
  rpc AddWorkerNode(AddWorkerNodeRequest) returns (AddWorkerNodeResponse);
  rpc ActivateWorkerNode(ActivateWorkerNodeRequest) returns (ActivateWorkerNodeResponse);
  rpc DeleteWorkerNode(DeleteWorkerNodeRequest) returns (DeleteWorkerNodeResponse);
  rpc ListAllNodes(ListAllNodesRequest) returns (ListAllNodesResponse);
  rpc GetClusterRecoveryStatus(GetClusterRecoveryStatusRequest) returns (GetClusterRecoveryStatusResponse);
  rpc GetMetaStoreInfo(GetMetaStoreInfoRequest) returns (GetMetaStoreInfoResponse);
}

enum SubscribeType {
  UNSPECIFIED = 0;
  FRONTEND = 1;
  HUMMOCK = 2;
  COMPACTOR = 3;
  COMPUTE = 4;
}

// Below for notification service.
message SubscribeRequest {
  SubscribeType subscribe_type = 1;
  common.HostAddress host = 2;
  uint32 worker_id = 3;
}

message MetaSnapshot {
  message SnapshotVersion {
    uint64 catalog_version = 1;
    reserved 2;
    reserved "parallel_unit_mapping_version";
    uint64 worker_node_version = 3;
    uint64 streaming_worker_slot_mapping_version = 4;
  }
  repeated catalog.Database databases = 1;
  repeated catalog.Schema schemas = 2;
  repeated catalog.Source sources = 3;
  repeated catalog.Sink sinks = 4;
  repeated catalog.Table tables = 5;
  repeated catalog.Index indexes = 6;
  repeated catalog.View views = 7;
  repeated catalog.Function functions = 15;
  repeated catalog.Connection connections = 17;
  repeated catalog.Subscription subscriptions = 19;
  repeated user.UserInfo users = 8;
  reserved 9;
  reserved "parallel_unit_mappings";
  GetSessionParamsResponse session_params = 20;
  repeated catalog.Secret secrets = 23;
  reserved 24;
  reserved "compute_node_total_cpu_count";
  common.ClusterResource cluster_resource = 25;
  repeated common.WorkerNode nodes = 10;
  hummock.HummockVersion hummock_version = 12;
  backup_service.MetaBackupManifestId meta_backup_manifest_id = 14;
  hummock.WriteLimits hummock_write_limits = 16;
  reserved 18;
  reserved "serving_parallel_unit_mappings";

  // for streaming
  repeated FragmentWorkerSlotMapping streaming_worker_slot_mappings = 21;
  repeated FragmentWorkerSlotMapping serving_worker_slot_mappings = 22;
  repeated ObjectDependency object_dependencies = 26;
  TableRefillRuntimeConfig table_refill_runtime_config = 27;

  SnapshotVersion version = 13;
}

message Object {
  oneof object_info {
    catalog.Database database = 1;
    catalog.Schema schema = 2;
    catalog.Table table = 3;
    catalog.Index index = 4;
    catalog.Source source = 5;
    catalog.Sink sink = 6;
    catalog.View view = 7;
    catalog.Function function = 8;
    catalog.Connection connection = 9;
    catalog.Subscription subscription = 10;
    catalog.Secret secret = 11;
  }
}

message ObjectGroup {
  repeated Object objects = 1;
  repeated ObjectDependency dependencies = 2;
}

message Recovery {}

message SubscribeResponse {
  enum Operation {
    UNSPECIFIED = 0;
    ADD = 1;
    DELETE = 2;
    UPDATE = 3;
    SNAPSHOT = 4;
  }
  common.Status status = 1;
  Operation operation = 2;

  // Catalog version
  uint64 version = 3;

  oneof info {
    catalog.Database database = 4;
    catalog.Schema schema = 5;
    catalog.Function function = 6;
    user.UserInfo user = 11;
    SetSessionParamRequest session_param = 26;
    common.WorkerNode node = 13;
    hummock.HummockVersionDeltas hummock_version_deltas = 15;
    MetaSnapshot snapshot = 16;
    backup_service.MetaBackupManifestId meta_backup_manifest_id = 17;
    SystemParams system_params = 19;
    hummock.WriteLimits hummock_write_limits = 20;
    ObjectGroup object_group = 21;
    catalog.Connection connection = 22;
    hummock.HummockVersionStats hummock_stats = 24;
    Recovery recovery = 25;
    FragmentWorkerSlotMapping streaming_worker_slot_mapping = 27;
    FragmentWorkerSlotMappings serving_worker_slot_mappings = 28;
    catalog.Secret secret = 29;
    common.ClusterResource cluster_resource = 31;
    TableRefillRuntimeConfig table_refill_runtime_config = 32;
  }
  reserved 12;
  reserved "parallel_unit_mapping";
  reserved 23;
  reserved "serving_parallel_unit_mappings";
  reserved 30;
  reserved "compute_node_total_cpu_count";
}

message TableRefillRuntimeConfig {
  TableCacheRefillPolicies table_cache_refill_policies = 1;
  ServingTableVnodeMappings serving_table_vnode_mappings = 2;
  uint64 version = 3;
}

message ServingTableVnodeMappings {
  repeated ServingTableVnodeMapping mappings = 1;
  message ServingTableVnodeMapping {
    uint32 table_id = 1;
    common.Buffer bitmap = 2;
  }
}

message TableCacheRefillPolicies {
  // Contains explicit result-table overrides only.
  repeated TableCacheRefillPolicy table_policies = 1;
  // Contains explicit internal-table overrides only.
  repeated TableCacheRefillPolicy internal_table_policies = 2;

  message TableCacheRefillPolicy {
    uint32 table_id = 1;
    CacheRefillPolicy policy = 2;

    enum CacheRefillPolicy {
      /// Unspecified policy, use global default configuration.
      UNSPECIFIED = 0;
      /// Enable normal cache refill for the table.
      ENABLED = 1;
      /// Disable cache refill for the table.
      DISABLED = 2;
      /// Enable cache refill optimized for streaming workloads for this table.
      STREAMING = 3;
      /// Enable cache refill optimized for serving workloads for this table.
      SERVING = 4;
      /// Enable cache refill optimized for both streaming and serving workloads for this table.
      BOTH = 5;
    }
  }
}

service NotificationService {
  rpc Subscribe(SubscribeRequest) returns (stream SubscribeResponse);
}

message GetClusterInfoRequest {}

message GetClusterInfoResponse {
  repeated common.WorkerNode worker_nodes = 1;
  repeated TableFragments table_fragments = 2;
  // `Source` and `SourceBackfill` are handled together here.
  map<uint32, source.ConnectorSplits> actor_splits = 3;
  map<uint32, catalog.Source> source_infos = 4;
  uint64 revision = 5;
}

// For each fragment that needs to be rescheduled, there will be a WorkerReschedule,
// indicating on which workers the actors of this fragment need to be changed and by how many.
message WorkerReschedule {
  // worker_id -> actor_diff
  map<uint32, int32> worker_actor_diff = 1;
}

message RescheduleRequest {
  reserved "reschedules";
  reserved 1;
  uint64 revision = 2;
  bool resolve_no_shuffle_upstream = 3;
  map<uint32, WorkerReschedule> worker_reschedules = 4;
}

message RescheduleResponse {
  bool success = 1;
  uint64 revision = 2;
}

message TableParallelism {
  message FixedParallelism {
    uint32 parallelism = 1;
  }

  // deprecated, treated as AdaptiveParallelism
  message AutoParallelism {}

  message AdaptiveParallelism {}

  message CustomParallelism {}

  oneof parallelism {
    FixedParallelism fixed = 1;
    AutoParallelism auto = 2;
    CustomParallelism custom = 3;
    AdaptiveParallelism adaptive = 4;
  }
}

// Changes a streaming job in place by overwriting its node_label.
// This may cause the re-scheduling of the streaming job actors.
message UpdateStreamingJobNodeLabelsRequest {
  // Id of the materialized view, table, or sink which we want to update
  uint32 id = 1;

  // replace the node_label of the streaming job with a given id with below value
  string node_label = 2;
}

// We do not need to add an explicit status field here, we can just use the RPC status
message UpdateStreamingJobNodeLabelsResponse {}

message GetServerlessStreamingJobsStatusRequest {}

// Descriptions of MVs and sinks
message GetServerlessStreamingJobsStatusResponse {
  message Status {
    uint32 table_id = 1;
    string node_label = 2;
    bool backfill_done = 3;
  }

  repeated Status streaming_job_statuses = 1;
}

// This is used by `risectl`
service ScaleService {
  rpc GetClusterInfo(GetClusterInfoRequest) returns (GetClusterInfoResponse);
  rpc Reschedule(RescheduleRequest) returns (RescheduleResponse);
  rpc UpdateStreamingJobNodeLabels(UpdateStreamingJobNodeLabelsRequest) returns (UpdateStreamingJobNodeLabelsResponse);
  rpc GetServerlessStreamingJobsStatus(GetServerlessStreamingJobsStatusRequest) returns (GetServerlessStreamingJobsStatusResponse);
}

message MembersRequest {}

message MetaMember {
  common.HostAddress address = 1;
  bool is_leader = 2;
}

message MembersResponse {
  repeated MetaMember members = 1;
}

service MetaMemberService {
  rpc Members(MembersRequest) returns (MembersResponse);
}

// The schema for persisted system parameters.
// Note on backward compatibility:
// - Do not remove deprecated fields. Mark them as deprecated instead.
// - Do not rename existing fields, since each field is stored separately in the meta store with the field name as the key.
// - To modify (rename, change the type or semantic of) a field, introduce a new field suffixed by the version.
message SystemParams {
  optional uint32 barrier_interval_ms = 1;
  optional uint64 checkpoint_frequency = 2;
  optional uint32 sstable_size_mb = 3;
  optional uint32 block_size_kb = 4;
  // Deprecated. Bloom filter is no longer a supported SST filter implementation; the field is
  // kept for backward compatibility and no longer controls whether SST filters are emitted.
  optional double bloom_false_positive = 5 [deprecated = true];
  optional string state_store = 6;
  optional string data_directory = 7;
  optional string backup_storage_url = 8;
  optional string backup_storage_directory = 9;
  // Deprecated. Use config file instead.
  optional bool telemetry_enabled = 10 [deprecated = true];
  optional uint32 parallel_compact_size_mb = 11;
  optional uint32 max_concurrent_creating_streaming_jobs = 12;
  optional bool pause_on_next_bootstrap = 13;
  optional string wasm_storage_url = 14 [deprecated = true];
  optional bool enable_tracing = 15;
  optional bool use_new_object_prefix_strategy = 16;
  optional string license_key = 17;
  optional uint64 time_travel_retention_ms = 18;
  optional string adaptive_parallelism_strategy = 19;
  optional bool per_database_isolation = 20;
  optional bool enforce_secret = 21;
}

message GetSystemParamsRequest {}

message GetSystemParamsResponse {
  SystemParams params = 1;
}

message SetSystemParamRequest {
  string param = 1;
  // None means set to default value.
  optional string value = 2;
}

message SetSystemParamResponse {
  optional SystemParams params = 1;
}

service SystemParamsService {
  rpc GetSystemParams(GetSystemParamsRequest) returns (GetSystemParamsResponse);
  rpc SetSystemParam(SetSystemParamRequest) returns (SetSystemParamResponse);
}

message GetSessionParamsRequest {}

message GetSessionParamsResponse {
  string params = 1;
}

message SetSessionParamRequest {
  string param = 1;
  // None means set to default value.
  optional string value = 2;
}

message SetSessionParamResponse {
  string param = 1;
}

// Used for alter system wide default parameters
service SessionParamService {
  rpc GetSessionParams(GetSessionParamsRequest) returns (GetSessionParamsResponse);
  rpc SetSessionParam(SetSessionParamRequest) returns (SetSessionParamResponse);
}

message GetServingVnodeMappingsRequest {}

message GetServingVnodeMappingsResponse {
  reserved 1;
  reserved "mappings";
  map<uint32, uint32> fragment_to_table = 2;
  repeated FragmentWorkerSlotMapping worker_slot_mappings = 3;
}

service ServingService {
  rpc GetServingVnodeMappings(GetServingVnodeMappingsRequest) returns (GetServingVnodeMappingsResponse);
}

message EventLog {
  message EventMetaNodeStart {
    string advertise_addr = 1;
    string listen_addr = 2;
    string opts = 3;
  }
  message EventCreateStreamJobFail {
    uint32 id = 1;
    string name = 2;
    string definition = 3;
    string error = 4;
  }
  message EventDirtyStreamJobClear {
    uint32 id = 1;
    string name = 2;
    string definition = 3;
    string error = 4;
  }
  message EventBarrierComplete {
    uint64 prev_epoch = 1;
    uint64 cur_epoch = 2;
    double duration_sec = 3;
    string command = 4;
    string barrier_kind = 5;
    uint32 database_id = 6;
  }
  message EventInjectBarrierFail {
    uint64 prev_epoch = 1;
    uint64 cur_epoch = 2;
    string error = 3;
  }
  message EventCollectBarrierFail {
    reserved 1, 2;
    reserved "prev_epoch", "cur_epoch";
    string error = 3;
  }
  message EventWorkerNodePanic {
    uint32 worker_id = 1;
    common.WorkerType worker_type = 2;
    common.HostAddress host_addr = 3;
    string panic_info = 4;
  }
  message EventAutoSchemaChangeFail {
    uint32 table_id = 1;
    string table_name = 2;
    string cdc_table_id = 3;
    string upstream_ddl = 4;
    string fail_info = 5;
  }
  message EventSinkFail {
    uint32 sink_id = 1;
    string sink_name = 2;
    string connector = 3;
    string error = 4;
  }
  message EventRecovery {
    message GlobalRecoveryStart {
      string reason = 1;
    }
    message GlobalRecoverySuccess {
      string reason = 1;
      float duration_secs = 2;
      repeated uint32 running_database_ids = 3;
      repeated uint32 recovering_database_ids = 4;
    }
    message GlobalRecoveryFailure {
      string reason = 1;
      string error = 2;
    }
    message DatabaseRecoveryStart {
      uint32 database_id = 1;
    }
    message DatabaseRecoveryFailure {
      uint32 database_id = 1;
    }
    message DatabaseRecoverySuccess {
      uint32 database_id = 1;
    }
    oneof recovery_event {
      GlobalRecoveryStart global_start = 1;
      GlobalRecoverySuccess global_success = 2;
      GlobalRecoveryFailure global_failure = 3;
      DatabaseRecoveryStart database_start = 4;
      DatabaseRecoveryFailure database_failure = 5;
      DatabaseRecoverySuccess database_success = 6;
    }
  }
  // Event logs identifier, which should be populated by event log service.
  optional string unique_id = 1;
  // Processing time, which should be populated by event log service.
  optional uint64 timestamp = 2;
  oneof event {
    EventCreateStreamJobFail create_stream_job_fail = 3;
    EventDirtyStreamJobClear dirty_stream_job_clear = 4;
    EventMetaNodeStart meta_node_start = 5;
    EventBarrierComplete barrier_complete = 6;
    EventInjectBarrierFail inject_barrier_fail = 7;
    EventCollectBarrierFail collect_barrier_fail = 8;
    EventLog.EventWorkerNodePanic worker_node_panic = 9;
    EventLog.EventAutoSchemaChangeFail auto_schema_change_fail = 10;
    EventLog.EventSinkFail sink_fail = 11;
    EventRecovery recovery = 12;
  }
}

message ListEventLogRequest {}

message ListEventLogResponse {
  repeated EventLog event_logs = 1;
}

message AddEventLogRequest {
  // A subset of EventLog.event that can be added by non meta node.
  oneof event {
    EventLog.EventWorkerNodePanic worker_node_panic = 1;
    EventLog.EventSinkFail sink_fail = 2;
    EventLog.EventAutoSchemaChangeFail auto_schema_change_fail = 3;
  }
}

message AddEventLogResponse {}

service EventLogService {
  rpc ListEventLog(ListEventLogRequest) returns (ListEventLogResponse);
  rpc AddEventLog(AddEventLogRequest) returns (AddEventLogResponse);
}

message ActorIds {
  repeated uint32 ids = 1;
}

message FragmentIdToActorIdMap {
  map<uint32, ActorIds> map = 1;
}

/// Provides all the ids: relation_id, fragment_id, actor_id
/// in an hierarchical format.
/// relation_id -> [fragment_id]
/// fragment_id -> [actor_id]
message RelationIdInfos {
  // relation_id -> FragmentIdToActorIdMap
  map<uint32, FragmentIdToActorIdMap> map = 1;
}

message FragmentToRelationMap {
  /// fragment_id -> relation_id mapping for all fragments
  map<uint32, uint32> fragment_to_relation_map = 1;
}

message ActorCountPerParallelism {
  message WorkerActorCount {
    uint64 actor_count = 1;
    uint64 parallelism = 2;
  }
  map<uint32, WorkerActorCount> worker_id_to_actor_count = 1;
  uint64 hard_limit = 2;
  uint64 soft_limit = 3;
}

message ClusterLimit {
  oneof limit {
    ActorCountPerParallelism actor_count = 1;
    // TODO: limit DDL using compaction pending bytes
  }
}

message GetClusterLimitsRequest {}

message GetClusterLimitsResponse {
  repeated ClusterLimit active_limits = 1;
}

service ClusterLimitService {
  rpc GetClusterLimits(GetClusterLimitsRequest) returns (GetClusterLimitsResponse);
}

message ListRateLimitsRequest {}

message ListRateLimitsResponse {
  message RateLimitInfo {
    uint32 fragment_id = 1;
    uint32 job_id = 2;
    uint32 fragment_type_mask = 3;
    uint32 rate_limit = 4;
    string node_name = 5;
  }
  repeated RateLimitInfo rate_limits = 1;
}

service HostedIcebergCatalogService {
  rpc ListIcebergTables(ListIcebergTablesRequest) returns (ListIcebergTablesResponse);
}

message ListIcebergTablesRequest {}

message ListIcebergTablesResponse {
  message IcebergTable {
    string catalog_name = 1;
    string table_namespace = 2;
    string table_name = 3;
    optional string metadata_location = 4;
    optional string previous_metadata_location = 5;
    optional string iceberg_type = 6;
  }
  repeated IcebergTable iceberg_tables = 1;
}

message SetSyncLogStoreAlignedRequest {
  uint32 job_id = 1;
  bool aligned = 2;
}

message SetSyncLogStoreAlignedResponse {}

message ListUnmigratedTablesRequest {}

message ListUnmigratedTablesResponse {
  message UnmigratedTable {
    uint32 table_id = 1;
    string table_name = 2;
  }
  repeated UnmigratedTable tables = 1;
}
