edition = "2023";
package grpc;

option go_package = "github.com/pbinitiative/zenbpm/internal/grpc/proto";

service ZenBpm {
  rpc JobStream(stream JobStreamRequest) returns (stream JobStreamResponse);
}

message JobStreamRequest{
  oneof request{
    StreamSubscriptionRequest subscription = 1;
    JobCompleteRequest complete = 2;
    JobFailRequest fail = 3;
    JobExtendLockRequest extend_lock = 4;
  }
}

message StreamSubscriptionRequest{
  enum Type {
    TYPE_UNDEFINED = 0; 
    TYPE_SUBSCRIBE = 1; // subscribe client to provided job_type
    TYPE_UNSUBSCRIBE = 2; // unsubscribe client from provided job_type
  }
  string job_type = 1;
  Type type = 2;
  // How long a job of this type stays locked for this client after delivery, in milliseconds.
  // 0 = the engine's default (jobManager.defaultLockDurationMs). A value above
  // jobManager.maxLockDurationMs is capped, never rejected; the effective deadline is
  // reported in lock_until of every delivery.
  int64 lock_duration_ms = 3;
  // How many jobs of this type this client may hold at once. 0 = the engine's default
  // (jobManager.defaultMaxActiveJobs). A value above jobManager.maxActiveJobsCap is capped.
  int32 max_active_jobs = 4;
}

message JobCompleteRequest{
  int64 key = 1; // job key
  bytes variables = 2; // json variables 
}

// Reports that the worker could not finish the job. With an error_code it is a BPMN
// error: a matching error catch event takes over and receives the variables, retries are
// not touched. Without one, an attempt is spent: while retries remain the job stays
// active, the variables are dropped and the job is handed out again after its backoff;
// at zero the job fails, keeps the variables as its output and an incident is created.
// retries and retry_backoff_ms are validated in every request: an invalid value is
// refused even next to an error_code, which then ignores them.
message JobFailRequest{
  int64 key = 1;
  string message = 2;
  string error_code = 3;
  bytes variables = 4;
  // Retries remaining after this failure. Absent = one less than now. Must not be
  // negative; a value above jobs.maxRetries is capped.
  int32 retries = 5;
  // How long the job waits before it is handed out again, in milliseconds. Absent =
  // the retryBackoff of the task definition, else jobs.defaultRetryBackoff; 0 = at once.
  // A value above jobs.maxRetryBackoff is capped.
  int64 retry_backoff_ms = 6;
  // The delivery the failure belongs to, WaitingJob.delivery_token. With it the failure is recorded
  // once: repeated, whatever it carries, it changes nothing and is answered as recorded. Only the
  // repeat of an error-coded failure which raised an incident is refused like a superseded one below
  // once that incident was resolved and a later delivery's failure was recorded: the failure history
  // keeps no entry naming its token. A failure of a delivery which was superseded, because the job
  // was handed out again after its lock lapsed or its leader changed, changes nothing either and is
  // refused with JOB_STREAM_ERROR_CODE_JOB_IN_TERMINAL_STATE: the newer delivery decides. A token
  // never handed out is refused with JOB_STREAM_ERROR_CODE_INVALID_REQUEST. Absent or 0, which no
  // delivery carries = every report counts, and one without an error code spends an attempt.
  int64 delivery_token = 7;
}

// Moves the lock deadline of a job this client holds to now plus lock_duration_ms.
// The engine answers with a LockExtended carrying the new deadline, or with an
// ErrorResult whose code is one of JobStreamErrorCode and a LockExtended carrying only the key.
message JobExtendLockRequest{
  int64 key = 1;
  // 0 = extend by the lock_duration_ms of the subscription the job was delivered under.
  // A value above jobManager.maxLockDurationMs is capped.
  int64 lock_duration_ms = 2;
  // The delivery whose lock to extend, WaitingJob.delivery_token. The lock of another
  // delivery of the job, such as the next one after this delivery's lock lapsed and the
  // job came back to this client, is not extended: JOB_STREAM_ERROR_CODE_LOCK_NOT_HELD.
  // Absent or 0, which no delivery carries = whatever delivery of the job this client holds.
  int64 delivery_token = 3;
}

message JobStreamResponse {
  ErrorResult error = 1;
  WaitingJob job = 2;
  // The answer to a JobExtendLockRequest. On an error it carries only the key.
  LockExtended lock_extended = 3;
}

message LockExtended {
  int64 key = 1;
  // Unix milliseconds on the clock of the partition leader at which the lock lapses.
  int64 lock_until = 2;
}

// Values of ErrorResult.code. Zero means the error carries no code.
enum JobStreamErrorCode {
  JOB_STREAM_ERROR_CODE_UNSPECIFIED = 0;
  // The lock lapsed, the job was completed or failed, it was never delivered to this stream, or no job has the key.
  JOB_STREAM_ERROR_CODE_LOCK_NOT_HELD = 1;
  // The job is currently locked for a different client id.
  JOB_STREAM_ERROR_CODE_LOCK_HELD_BY_OTHER_CLIENT = 2;
  // The leader of the job's partition could not be reached or has just changed. The
  // outcome is unconfirmed: an extension may already have moved the deadline, and a
  // leader change forgets the lock, so retry the extension in a moment; a failure may
  // already have been recorded: repeating one which names its delivery token changes nothing
  // more, and one without a token spends another attempt, or, with an error code, is refused
  // once the first was recorded.
  JOB_STREAM_ERROR_CODE_LEADER_UNAVAILABLE = 3;
  // The request itself is wrong, such as a fail request with negative retries or a
  // negative backoff; nothing was changed and repeating it gets the same answer.
  JOB_STREAM_ERROR_CODE_INVALID_REQUEST = 4;
  // No job has this key: it never existed, or its instance was deleted.
  JOB_STREAM_ERROR_CODE_JOB_NOT_FOUND = 5;
  // The job no longer waits for this delivery: it was completed, terminated or already
  // failed, or, for a failure naming its delivery token, it was handed out again after
  // that delivery. Nothing was changed.
  JOB_STREAM_ERROR_CODE_JOB_IN_TERMINAL_STATE = 6;
}

message WaitingJob {
  int64 key = 1;
  int64 instance_key = 2;
  bytes input_variables = 3;
  string type = 4;
  string element_id = 5;
  int64 created_at = 6;
  string element_type = 7;
  // Unix milliseconds on the clock of the partition leader at which the lock this
  // delivery holds lapses. With synchronised clocks, lock_until minus the local time
  // estimates the remaining time; it is conservative, the leader counts the lock from
  // the end of the send. When the clocks cannot be trusted, count the lock duration
  // of the subscription from the moment of receipt instead of comparing timestamps.
  int64 lock_until = 8;
  map<string, string> headers = 9;
  // Attempts left: failures without an error code the job may still report; the one which leaves none creates an incident.
  int32 retries = 10;
  // 1 for the first delivery of a series of attempts, one more after every failure without an
  // error code; two deliveries of one attempt (a lapsed lock) carry the same number. It tells
  // a redelivery of an attempt from a retry within one series; resolving an incident starts a
  // new series at 1, so (key, attempt) is not unique over the life of a job. Key an effect
  // which must happen once per job on the job key, or on a key of the business operation.
  int32 attempt = 11;
  // Identifies this delivery: every delivery of the job carries a higher token than the one
  // before, a redelivery of the same attempt after a lapsed lock or a leader change included.
  // Send it back as JobFailRequest.delivery_token. Absent when the leader which handed the job
  // out issues no tokens, as during a rolling upgrade; the failure then names none, and a client
  // which copies the field whether or not it is present sends 0, which names none as well.
  int64 delivery_token = 12;
}

message ErrorResult {
  uint32 code = 1;
  string message = 2;
}