syntax = "proto3";

package nominal.direct_channel_writer.v2;

import "google/api/annotations.proto";
import "nominal/protos/gen/v1/alias.proto";
import "nominal/protos/gen/v1/error.proto";
import "nominal/protos/types/time/time.proto";

option java_multiple_files = true;
option java_package = "io.nominal.direct_channel_writer.v2";

// Public-facing write request. Used by external clients via the Conjure HTTP API.
message WriteBatchesRequest {
  repeated RecordsBatch batches = 1;
  string data_source_rid = 2 [(nominal.gen.v1.alias) = {java_name: "io.nominal.api.rids.NominalDataSourceOrDatasetRid"}];
}

message StreamingSessionSourceInfo {
  oneof source {
    MeshStreamingSessionSourceInfo mesh = 1;
    DataConnectorStreamingSessionSourceInfo data_connector = 2;
    CustomStreamingSessionSourceInfo custom = 3;
  }
}

message MeshStreamingSessionSourceInfo {
  string source_dataset_rid = 1 [(nominal.gen.v1.alias) = {java_name: "io.nominal.api.rids.DatasetRid"}];
  string source_org_rid = 2 [(nominal.gen.v1.alias) = {java_name: "io.nominal.authentication.api.OrgRid"}];
  string origin_streaming_session_source = 3;
}

message DataConnectorStreamingSessionSourceInfo {
  string data_connector_rid = 1 [(nominal.gen.v1.alias) = {java_name: "io.nominal.api.rids.DataConnectorRid"}];
}

message CustomStreamingSessionSourceInfo {
  string source_name = 1;
}

// Internal write request. Used internally by the channel writer, Kafka consumers, and mesh.
message InternalWriteBatchesRequest {
  repeated RecordsBatch batches = 1;
  string data_source_rid = 2 [(nominal.gen.v1.alias) = {java_name: "io.nominal.api.rids.NominalDataSourceOrDatasetRid"}];
  StreamingSessionSourceInfo source = 3;
  // Streaming session RID resolved during the original write attempt, embedded before enqueueing
  // to the retry topic so CW can use it directly on retry without re-resolving against a potentially
  // expired session cache.
  string streaming_session_rid = 4 [(nominal.gen.v1.alias) = {java_name: "io.nominal.ingest.api.StreamingSessionRid"}];
}

message ArrayPoints {
  oneof array_type {
    DoubleArrayPoints double_array_points = 1;
    StringArrayPoints string_array_points = 2;
  }
}

message StringArrayPoints {
  repeated StringArrayPoint points = 1;
}

message StringArrayPoint {
  repeated string value = 2;
}

message DoubleArrayPoints {
  repeated DoubleArrayPoint points = 1;
}

message DoubleArrayPoint {
  repeated double value = 2;
}

message LogPoints {
  repeated LogPoint points = 1;
}

message LogPoint {
  LogValue value = 2;
}

message StructPoints {
  repeated string points = 1;
}

message LogValue {
  string message = 1;

  map<string, string> args = 2;
}

message StringPoints {
  repeated string points = 1;
}

message DoublePoints {
  repeated double points = 1;
}

message IntPoints {
  repeated int64 points = 1;
}

message Uint64Points {
  repeated uint64 points = 1;
}

message Points {
  repeated nominal.types.time.Timestamp timestamps = 1;
  oneof points {
    DoublePoints double_points = 2;
    StringPoints string_points = 3;
    LogPoints log_points = 4;
    IntPoints int_points = 5;
    ArrayPoints array_points = 6;
    StructPoints struct_points = 7;
    Uint64Points uint64_points = 8;
  }
}

message RecordsBatch {
  string channel = 1 [(nominal.gen.v1.alias) = {java_name: "io.nominal.api.Channel"}];

  map<string, string> tags = 2 [(nominal.gen.v1.alias) = {
    key_java_name: "io.nominal.api.TagName"
    value_java_name: "io.nominal.api.TagValue"
  }];

  Points points = 3;
}

message ChannelSeriesMetadata {
  string series_id = 1;

  map<string, string> tags = 2 [(nominal.gen.v1.alias) = {
    key_java_name: "io.nominal.api.TagName"
    value_java_name: "io.nominal.api.TagValue"
  }];
}

message ChannelSeriesMetadataBatch {
  repeated ChannelSeriesMetadata series_metadata = 1;
}

message WriteBatchesResponse {}

message WriteRetryBatchesResponse {}

// buf:lint:ignore ENUM_ZERO_VALUE_SUFFIX
enum NominalDirectChannelWriterError {
  NOMINAL_DIRECT_CHANNEL_WRITER_ERROR_DATA_SOURCE_NOT_FOUND = 0 [(nominal.gen.v1.error) = {
    code: "NOT_FOUND"
    message: "DataSourceNotFound"
  }];
  NOMINAL_DIRECT_CHANNEL_WRITER_ERROR_DATA_SOURCES_NOT_FOUND = 1 [(nominal.gen.v1.error) = {
    code: "NOT_FOUND"
    message: "DataSourcesNotFound"
  }];
  NOMINAL_DIRECT_CHANNEL_WRITER_ERROR_INVALID_NOMINAL_DATA_SOURCE = 2 [(nominal.gen.v1.error) = {
    code: "INVALID_ARGUMENT"
    message: "InvalidNominalDataSource"
  }];
  // The requested data source already exists.
  NOMINAL_DIRECT_CHANNEL_WRITER_ERROR_NOMINAL_DATA_SOURCE_CONFLICT = 3 [(nominal.gen.v1.error) = {
    code: "ALREADY_EXISTS"
    message: "The requested data source already exists."
  }];

  // Array size exceeds maximum allowed length.
  NOMINAL_DIRECT_CHANNEL_WRITER_ERROR_ARRAY_TOO_LARGE = 4 [(nominal.gen.v1.error) = {
    code: "INVALID_ARGUMENT"
    message: "Array size exceeds maximum allowed length."
  }];
  // Found two different data types for the same series in the request.
  NOMINAL_DIRECT_CHANNEL_WRITER_ERROR_CONFLICTING_DATA_TYPES = 5 [(nominal.gen.v1.error) = {
    code: "INVALID_ARGUMENT"
    message: "Found two different data types for the same series in the request."
  }];
  // The nominal data source or dataset provided is invalid.
  NOMINAL_DIRECT_CHANNEL_WRITER_ERROR_INVALID_DATA_SOURCE = 6 [(nominal.gen.v1.error) = {
    code: "INVALID_ARGUMENT"
    message: "The nominal data source or dataset provided is invalid."
  }];
  NOMINAL_DIRECT_CHANNEL_WRITER_ERROR_INVALID_TELEGRAF_TIMESTAMP = 7 [(nominal.gen.v1.error) = {
    code: "INVALID_ARGUMENT"
    message: "InvalidTelegrafgoogle.protobuf.Timestamp"
  }];
  NOMINAL_DIRECT_CHANNEL_WRITER_ERROR_INVALID_TIMESTAMP = 8 [(nominal.gen.v1.error) = {
    code: "INVALID_ARGUMENT"
    message: "Invalidgoogle.protobuf.Timestamp"
  }];
  // Columnar streaming request contains different numbers of timestamps and values
  NOMINAL_DIRECT_CHANNEL_WRITER_ERROR_MISMATCHED_TIMESTAMPS_AND_VALUES = 9 [(nominal.gen.v1.error) = {
    code: "INVALID_ARGUMENT"
    message: "Columnar streaming request contains different numbers of timestamps and values"
  }];
  // Unsupported dataset for streaming.
  NOMINAL_DIRECT_CHANNEL_WRITER_ERROR_STREAMING_DISABLED_ON_DATASET = 10 [(nominal.gen.v1.error) = {
    code: "INVALID_ARGUMENT"
    message: "Unsupported dataset for streaming."
  }];
  // The provided Range is invalid because the start time is greater than the end time.
  NOMINAL_DIRECT_CHANNEL_WRITER_ERROR_INVALID_RANGE = 11 [(nominal.gen.v1.error) = {
    code: "INVALID_ARGUMENT"
    message: "The start time of the provided Range is greater than its end time."
  }];
  // The provided Points type is not set.
  NOMINAL_DIRECT_CHANNEL_WRITER_ERROR_POINTS_TYPE_NOT_SET = 12 [(nominal.gen.v1.error) = {
    code: "INVALID_ARGUMENT"
    message: "The provided ArrayPoints type or Points type is not set."
  }];
  // The provided ArrayPoints type is not set.
  NOMINAL_DIRECT_CHANNEL_WRITER_ERROR_ARRAY_POINTS_TYPE_NOT_SET = 13 [(nominal.gen.v1.error) = {
    code: "INVALID_ARGUMENT"
    message: "The provided ArrayPoints type is not set."
  }];
  // Not authorized to write to the specified data source.
  NOMINAL_DIRECT_CHANNEL_WRITER_ERROR_NOT_AUTHORIZED = 14 [(nominal.gen.v1.error) = {
    code: "PERMISSION_DENIED"
    message: "Not authorized to write to the specified data source."
  }];
}

// Writes data points directly to Nominal's managed database offering.
service DirectNominalChannelWriterService {
  // buf:lint:ignore RPC_REQUEST_STANDARD_NAME
  // Synchronously writes batches of records to a Nominal data source. This endpoint bypasses the
  // Channel Writer service entirely, and should only be used if the implications are well understood.
  //
  // If you call this endpoint, writes will go directly into Nominal DB and will not be placed in Nominal's
  // durable queue. This results in lower latency, but also consequently lower durability.
  rpc WriteBatches(InternalWriteBatchesRequest) returns (WriteBatchesResponse) {
    option (google.api.http) = {
      post: "/storage/direct-writer/v2"
      body: "*"
    };
  }
}
