syntax = "proto3";

package nominal.file_ingest.v1;

import "nominal/protos/direct_channel_writer/v2/file_data.proto";
import "nominal/protos/file_ingest/v1/ingest_request.proto";
import "nominal/protos/gen/v1/alias.proto";
import "nominal/protos/types/object_storage/handle.proto";
import "nominal/protos/types/time/time.proto";

option java_multiple_files = true;
option java_package = "io.nominal.file_ingest.v1";

// Wire types for refinery's Temporal activities. Each activity request wraps
// the inputs needed for one operation; responses are intentionally empty
// today so the contract can grow without rewriting activity signatures.

message SetDatasetFileIngestStatusRequest {
  string dataset_file_id = 1 [(nominal.gen.v1.alias) = {java_name: "io.nominal.datasource.DatasetFileId"}];
  string dataset_rid = 2 [(nominal.gen.v1.alias) = {java_name: "io.nominal.api.rids.DatasetRid"}];

  // Refinery only drives parsing, ingesting, and error transitions; other IngestStatusV2 variants
  // are set elsewhere. The empty payloads exist so we can grow them later (e.g.
  // Parsing.bytes_processed) without changing the activity signature.
  oneof status {
    Parsing parsing = 3;
    Ingesting ingesting = 4;
    Error error = 5;
  }

  string ingest_job_rid = 6 [(nominal.gen.v1.alias) = {java_name: "io.nominal.ingest.api.IngestJobRid"}];
}

message SetDatasetFileIngestStatusResponse {}

message Parsing {}
message Ingesting {
  nominal.types.time.Range bounds = 1;
}

message Error {
  string error_type = 1;
  string message = 2;
  string org_rid = 3 [(nominal.gen.v1.alias) = {java_name: "io.nominal.authentication.api.OrgRid"}];
  string file_type = 4;
}

message ParseFileRequest {
  string ingest_job_rid = 1 [(nominal.gen.v1.alias) = {java_name: "io.nominal.ingest.api.IngestJobRid"}];
  string dataset_file_id = 2 [(nominal.gen.v1.alias) = {java_name: "io.nominal.datasource.DatasetFileId"}];
  string dataset_rid = 3 [(nominal.gen.v1.alias) = {java_name: "io.nominal.api.rids.DatasetRid"}];
  string workspace_rid = 5 [(nominal.gen.v1.alias) = {java_name: "io.nominal.api.rids.WorkspaceRid"}];
  nominal.types.object_storage.Handle handle = 6;
  TimestampMetadata timestamp_metadata = 7;
  map<string, string> additional_tags = 8 [(nominal.gen.v1.alias) = {
    key_java_name: "io.nominal.api.TagName"
    value_java_name: "io.nominal.api.TagValue"
  }];

  // Mirrors IngestFileRequest.ingest so the factory can dispatch on log vs data at the
  // top of the request rather than relying on hasDataIngest() / hasLogIngest().
  oneof ingest {
    LogFileIngest log_ingest = 9;
    DataFileIngest data_ingest = 10;
  }
}

// Parses the workflow input into staged object-storage batches.
message ParseFileResponse {
  repeated StagedBatch staged_batches = 1;
  nominal.types.time.Range bounds = 2;
}

// Publishes one WriteFileDataRequest to Kafka per staged batch, keyed by dataset_rid.
// Each batch's handle points at the staged payload in object storage.
message WriteFileBatchesToKafkaRequest {
  repeated StagedBatch staged_batches = 1;
  string ingest_job_rid = 2 [(nominal.gen.v1.alias) = {java_name: "io.nominal.ingest.api.IngestJobRid"}];
  string file_rid = 3;
  string org_rid = 4 [(nominal.gen.v1.alias) = {java_name: "io.nominal.authentication.api.OrgRid"}];
  string dataset_file_id = 5 [(nominal.gen.v1.alias) = {java_name: "io.nominal.datasource.DatasetFileId"}];
  string dataset_rid = 6 [(nominal.gen.v1.alias) = {java_name: "io.nominal.api.rids.DatasetRid"}];
}

message StagedBatch {
  int32 batch_id = 1;
  nominal.types.object_storage.Handle handle = 2;
  nominal.direct_channel_writer.v2.BatchFormat format = 3;
}

message WriteFileBatchesToKafkaResponse {}
