syntax = "proto3";

package nominal.ingest.v1.internal;

import "buf/validate/validate.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.ingest.v1.internal";

message MarkBatchesCompletedRequest {
  string ingest_job_rid = 1 [
    (buf.validate.field).required = true,
    (nominal.gen.v1.alias) = {java_name: "io.nominal.ingest.api.IngestJobRid"}
  ];
  string dataset_file_id = 2 [
    (buf.validate.field).required = true,
    (nominal.gen.v1.alias) = {java_name: "io.nominal.datasource.DatasetFileId"}
  ];
  // The batches to mark COMPLETED, each carrying the object-storage location of its staged
  // payload. The location is recorded on the batch row at completion time so the retention
  // sweep can reclaim the staged object; it is carried here rather than on
  // CreatePendingBatchesRequest because completion calls are bounded by the consumer's poll
  // size, while a large file's full batch list is not bounded by anything.
  repeated CompletedBatch batches = 3 [(buf.validate.field).repeated.min_items = 1];
}

message CompletedBatch {
  int32 batch_id = 1;
  nominal.types.object_storage.Handle handle = 2 [(buf.validate.field).required = true];
}

message MarkBatchesCompletedResponse {}

message CreatePendingBatchesRequest {
  string ingest_job_rid = 1 [
    (buf.validate.field).required = true,
    (nominal.gen.v1.alias) = {java_name: "io.nominal.ingest.api.IngestJobRid"}
  ];
  string dataset_file_id = 2 [
    (buf.validate.field).required = true,
    (nominal.gen.v1.alias) = {java_name: "io.nominal.datasource.DatasetFileId"}
  ];
  // Number of batches expected for this file; must be positive. Deliberately a count rather
  // than a per-batch list: a large file can stage tens of thousands of batches, more than a
  // single request should carry. Each batch's staged location is recorded later, on the
  // poll-bounded MarkBatchesCompleted call.
  int32 num_batches = 3 [(buf.validate.field).int32.gt = 0];
}

message CreatePendingBatchesResponse {}

message SetDatasetFileIngestStatusRequest {
  string ingest_job_rid = 1 [
    (buf.validate.field).required = true,
    (nominal.gen.v1.alias) = {java_name: "io.nominal.ingest.api.IngestJobRid"}
  ];
  string dataset_file_id = 2 [
    (buf.validate.field).required = true,
    (nominal.gen.v1.alias) = {java_name: "io.nominal.datasource.DatasetFileId"}
  ];
  // The transition to apply. Refinery drives only parsing, ingesting, and error.
  oneof status {
    Parsing parsing = 3;
    Ingesting ingesting = 4;
    Error error = 5;
  }
}

message SetDatasetFileIngestStatusResponse {}

message Parsing {}

message Ingesting {
  // Timestamp bounds observed while parsing the file, applied to the dataset file row.
  nominal.types.time.Range bounds = 1;
}

message Error {
  string error_type = 1;
  string message = 2;
}

// Internal service for inter-service RPCs. Uses a separate package so that
// the gRPC path (nominal.ingest.v1.internal.*) is excluded from public ingress.
service InternalIngestBatchService {
  // Mark the supplied batches for {@code dataset_file_id} as COMPLETED. Within the same
  // transaction, if all batches for the file are now complete, mark the file's ingest
  // status as COMPLETED; if all files for {@code ingest_job_rid} are then in a terminal
  // ingest status (COMPLETED or FAILED), mark the ingest job COMPLETED. Internal-only
  // (service-token auth).
  //
  // Idempotent: re-issuing the same request after success is a no-op. Already-COMPLETED
  // batches/files/jobs are left in place and the cascade short-circuits at each step.
  rpc MarkBatchesCompleted(MarkBatchesCompletedRequest) returns (MarkBatchesCompletedResponse);

  // Set a dataset file's ingest status (PARSING / INGESTING / ERROR) during the refinery file
  // workflow. On an ERROR transition the owning ingest job is locked and, if all of its files are
  // now terminal, marked COMPLETED (if any file succeeded) or FAILED (if all failed); PARSING /
  // INGESTING transitions only update the file row. Internal-only (service-token auth).
  rpc SetDatasetFileIngestStatus(SetDatasetFileIngestStatusRequest) returns (SetDatasetFileIngestStatusResponse);

  // Insert PENDING batch rows for {@code dataset_file_id} before its batches are published.
  // Idempotent on retry. Internal-only (service-token auth).
  rpc CreatePendingBatches(CreatePendingBatchesRequest) returns (CreatePendingBatchesResponse);
}
