api.proto

syntax = "proto3";
package proto;

option java_package = "io.liftbridge.proto";


message NullableInt64 {
    int64 value = 1; 
}

message NullableInt32 {
    int32 value = 1; 
}

message NullableBool {
    bool value = 1; 
}

// CreateStreamRequest is sent to create a new stream.
message CreateStreamRequest {
    string        subject                       = 1;  // Stream NATS subject
    string        name                          = 2;  // Stream name (unique per subject)
    string        group                         = 3;  // Partitions NATS subject amongst group members
    int32         replicationFactor             = 4;  // Number of stream replicas
    int32         partitions                    = 5;  // Number of stream partitions
    NullableInt64 retentionMaxBytes             = 6;  // The maximum size a stream's log can grow to, in bytes
    NullableInt64 retentionMaxMessages          = 7;  // The maximum size a stream's log can grow to, in messages
    NullableInt64 retentionMaxAge               = 8;  // The TTL for stream log segment files 
    NullableInt64 cleanerInterval               = 9;  // The frequency to check for log cleaner
    NullableInt64 segmentMaxBytes               = 10; // The maximum size of a single stream log segment file in bytes
    NullableInt64 segmentMaxAge                 = 11; // The maximum time before a new stream log segment is rolled out
    NullableInt32 compactMaxGoroutines          = 12; // The maximum number of concurrent goroutines to use for compaction on a stream log
    NullableBool  compactEnabled                = 13; // CompactEnabled controls compaction for a stream
    NullableInt64 autoPauseTime                 = 14; // The amount of time a stream partition can go idle before it is automatically paused
    NullableBool  autoPauseDisableIfSubscribers = 15; // Disables automatic partition pausing when there are subscribers
    NullableInt32 minIsr                        = 16; // The minimum number of replicas that must ack a stream write before it can be committed
    NullableBool  optimisticConcurrencyControl  = 17; // Set activation mode for Optimistic Concurrency Control on stream. Default is false
    NullableBool  encryption                    = 18; // Set activation mode for encryption of data-at-rest on stream. Default is false

}

// CreateStreamResponse is sent by server after creating a stream.
message CreateStreamResponse {
    // Intentionally empty.
}

// DeleteStreamRequest is sent to delete a stream.
message DeleteStreamRequest {
    string name              = 1; // Stream name
}

// DeleteStreamResponse is sent by server after deleting a stream.
message DeleteStreamResponse {
    // Intentionally empty.
}

// PauseStreamRequest is sent to pause the specified stream partitions. If no
// partitions are specified, this will pause all partitions in the stream.
// Partitions are resumed when they are published to via the Liftbridge Publish
// API.
message PauseStreamRequest {
    string         name       = 1; // Stream name
    repeated int32 partitions = 2; // Stream partitions
    bool           resumeAll  = 3; // Resume all partitions at once
}

// PauseStreamResponse is sent by server after pausing a stream.
message PauseStreamResponse {
    // Intentionally empty.
}

// SetStreamReadonlyRequest is send to set a stream as read-only.
message SetStreamReadonlyRequest {
    string name               = 1; // Stream name
    repeated int32 partitions = 2; // Stream partitions
    bool readonly             = 3; // Should the readonly flag be set or removed
}

// SetStreamReadonlyResponse is sent by server after setting a stream's readonly
// flag.
message SetStreamReadonlyResponse {
    // Intentionally empty.
}

// StartPosition determines the start-position type on a subscription.
enum StartPosition {
    NEW_ONLY    = 0; // Start at new messages after the latest
    OFFSET      = 1; // Start at a specified offset
    EARLIEST    = 2; // Start at the oldest message
    LATEST      = 3; // Start at the newest message
    TIMESTAMP   = 4; // Start at a specified timestamp
}

// StopPosition determines the stop-position type on a subscription.
enum StopPosition {
    STOP_ON_CANCEL = 0; // Stop when the request is cancelled
    STOP_OFFSET    = 1; // Stop at a specified offset
    STOP_LATEST    = 2; // Stop at the latest offset
    STOP_TIMESTAMP = 3; // Stop at a specified timestamp
}

// Consumer contains information pertaining to a subscriber that is a member of
// a consumer group.
message Consumer {
    string groupId    = 1; // ID of group consumer is a member of
    uint64 groupEpoch = 2; // Consumer group generation for consumer fencing
    string consumerId = 3; // Unique consumer ID
}

// SubscribeRequest is sent to subscribe to a stream partition.
message SubscribeRequest {
    string        stream         = 1; // Stream name to subscribe to
    int32         partition      = 2; // Stream partition to subscribe to
    StartPosition startPosition  = 3; // Where to begin consuming from
    int64         startOffset    = 4 [jstype=JS_STRING]; // Offset to begin consuming from
    int64         startTimestamp = 5 [jstype=JS_STRING]; // Timestamp to begin consuming from
    bool          readISRReplica = 6; // Subscribe from a random ISR replica instead of choosing explicitly leader
    bool          resume         = 7; // Allow resuming a paused partition before subscribing to it
    StopPosition  stopPosition   = 8; // Where to stop consuming
    int64         stopOffset     = 9  [jstype=JS_STRING]; // Offset to stop consuming at
    int64         stopTimestamp  = 10 [jstype=JS_STRING]; // Timestamp to stop consuming at
    Consumer      consumer       = 11; // Set if subscriber is part of a consumer group
}

// FetchMetadataRequest is sent to retrieve the latest cluster metadata.
message FetchMetadataRequest {
    repeated string streams = 1; // The streams to fetch metadata for (all if empty)
    repeated string groups  = 2; // The consumer groups to fetch metadata for
}

// FetchMetadataResponse contains the cluster metadata requested.
message FetchMetadataResponse {
    repeated Broker                brokers        = 1; // Information for all brokers
    repeated StreamMetadata        streamMetadata = 2; // Information for requested streams
    repeated ConsumerGroupMetadata groupMetadata  = 3; // Information for requested groups
}

// FetchPartitionMetadataRequest is sent to retrieve latest partition metadata.
// Mainly it is useful to retrieve Highest Watermark and Highest Offset of the partition leader
message FetchPartitionMetadataRequest {
    string stream       = 1;
    int32  partition    = 2;
}


//  FetchPartitionMetadataResponse contains the partition metadata requested
message FetchPartitionMetadataResponse {
    PartitionMetadata metadata = 1;
}

// PublishRequest is sent to publish a new message to a stream.
message PublishRequest {
    bytes               key             = 1; // Message key
    bytes               value           = 2; // Message payload
    string              stream          = 3; // Stream name to publish to
    int32               partition       = 4; // Stream partition to publish to
    map<string, bytes>  headers         = 5; // Message headers
    string              ackInbox        = 6; // NATS subject to publish acks to
    string              correlationId   = 7; // User-supplied value to correlate acks to publishes
    AckPolicy           ackPolicy       = 8; // Controls the behavior of acks
    int64               expectedOffset  = 9 [jstype=JS_STRING]; // Provide expected offset on the partion after publish
}

// PublishAsyncError is returned on the PublishAsync stream if a publish failed.
message PublishAsyncError {
    enum Code {
        UNKNOWN           = 0;
        BAD_REQUEST       = 1; // Invalid request arguments
        NOT_FOUND         = 2; // Stream or partition not found
        INTERNAL          = 3; // Internal server error
        READONLY          = 4; // Partition is readonly
        INCORRECT_OFFSET  = 5; // Publish is sent with incorrect expected offset. Used for concurrency control
        ENCRYPTION_FAILED = 6; // Failed to encrypt data on partition
        PERMISSION_DENIED = 7; // Fail to perform action on resource.
    }
    Code   code    = 1; // Error code
    string message = 2; // Error message
}

// PublishResponse is sent by the server after publishing a message to a
// stream.
message PublishResponse {
    Ack               ack           = 1; // The ack for the published message if AckPolicy was not NONE
    PublishAsyncError asyncError    = 2; // Set if publish failed (only set on PublishAsync)
    string            correlationId = 3; // User-supplied value to correlate async responses to publishes
}

// PublishToSubjectRequest is sent to publish a Liftbridge message to a NATS
// subject.
message PublishToSubjectRequest {
    bytes               key           = 1; // Message key
    bytes               value         = 2; // Message payload
    string              subject       = 3; // NATS subject to publish to
    map<string, bytes>  headers       = 4; // Message headers
    string              ackInbox      = 5; // NATS subject to publish acks to
    string              correlationId = 6; // User-supplied value to correlate acks to publishes
    AckPolicy           ackPolicy     = 7; // Controls the behavior of acks
}

// PublishToSubjectResponse is sent by the server after publishing a message to
// a NATS subject.
message PublishToSubjectResponse {
    Ack ack = 1; // The ack for the published message if AckPolicy was not NONE
}

// SetCursorRequest is sent to set a consumer's cursor position for a
// particular stream partition.
message SetCursorRequest {
    string stream    = 1; // Stream name
    int32  partition = 2; // Stream partition
    string cursorId  = 3; // User-supplied value to uniquely identify this cursor
    int64  offset    = 4 [jstype=JS_STRING]; // Offset where the cursor should be placed
}

// SetCursorResponse is sent by the server after setting a consumer cursor.
message SetCursorResponse {
    // Intentionally empty
}

// FetchCursorRequest is sent to retrieve a consumer's cursor position for a
// particular stream partition.
message FetchCursorRequest {
    string stream    = 1; // Stream name
    int32  partition = 2; // Stream partition
    string cursorId  = 3; // User-supplied value to uniquely identify this cursor
}

// FetchCursorResponse contains the requested consumer cursor position.
message FetchCursorResponse {
    int64 offset = 1 [jstype=JS_STRING]; // Offset where cursor is located
}

// JoinConsumerGroupRequest is sent to add a consumer to a consumer group.
message JoinConsumerGroupRequest {
    string          groupId    = 1; // ID of consumer group to join
    string          consumerId = 2; // Uniquely identifies the consumer
    repeated string streams    = 3; // Streams to consume
}

// JoinConsumerGroupResponse is sent by the server after adding a consumer to a
// consumer group.
message JoinConsumerGroupResponse {
    string coordinator        = 1; // Broker acting as consumer group coordinator
    uint64 epoch              = 2 [jstype=JS_STRING]; // Consumer group epoch for fencing coordinator generations
    int64  consumerTimeout    = 3 [jstype=JS_STRING]; // Coordinator's consumer liveness check frequency
    int64  coordinatorTimeout = 4 [jstype=JS_STRING]; // Timeout in which consumer should report coordinator as failed
}

// LeaveConsumerGroupRequest is sent to remove a consumer from a consumer
// group.
message LeaveConsumerGroupRequest {
    string groupId    = 1; // ID of consumer group to leave
    string consumerId = 2; // Consumer to remove
}

// LeaveConsumerGroupResponse is sent by the server after removing a consumer
// from a consumer group.
message LeaveConsumerGroupResponse {
    // Intentionally empty.
}

// FetchConsumerGroupAssignmentsRequest is sent to a consumer group coordinator
// to retrieve the consumer's partition assignments.
message FetchConsumerGroupAssignmentsRequest {
    string groupId    = 1;
    string consumerId = 2;
    uint64 epoch      = 3 [jstype=JS_STRING];
}

// PartitionAssignment contains a set of stream partitions that are assigned to
// a consumer.
message PartitionAssignment {
    string         stream     = 1;
    repeated int32 partitions = 2;
}

// FetchConsumerGroupAssignmentsResponse is sent by the group coordinator to
// notify the consumer of its partition assignments.
message FetchConsumerGroupAssignmentsResponse {
    uint64                       epoch       = 1 [jstype=JS_STRING]; // Group generation for consumer fencing
    repeated PartitionAssignment assignments = 2; // Partitions assigned to consumer
}

// ReportConsumerGroupCoordinatorRequest is sent by a consumer group member to
// report the group coordinator as failed.
message ReportConsumerGroupCoordinatorRequest {
    string groupId     = 1;
    string consumerId  = 2;
    string coordinator = 3;
    uint64 epoch       = 4 [jstype=JS_STRING];
}

// ReportConsumerGroupCoordinatorResponse is sent by the server in response to
// a request to report a consumer group coordinator.
message ReportConsumerGroupCoordinatorResponse {
    // Intentionally empty
}

// Broker contains information for a Liftbridge broker.
message Broker {
    string id             = 1; // Broker id
    string host           = 2; // Broker host
    int32  port           = 3; // Broker port
    int32  partitionCount = 4; // Number of partitions stored on broker
    int32  leaderCount    = 5; // Number of partitions stored on broker as partition leader
}

// StreamMetadata contains information for a stream.
message StreamMetadata {
    enum Error {
        OK             = 0;
        UNKNOWN_STREAM = 1;
    }
    string                        name              = 1; // The name of the stream being described
    string                        subject           = 2; // The stream subject
    Error                         error             = 3; // Indicates if there was something wrong with the requested stream
    map<int32, PartitionMetadata> partitions        = 4; // Information for the stream partitions
    int64                         creationTimestamp = 5 [jstype=JS_STRING]; // When the stream has been created
}

// ConsumerGroupMetadata contains information for a consumer group.
message ConsumerGroupMetadata {
    enum Error {
        OK            = 0;
        UNKNOWN_GROUP = 1;
    }
    string groupId     = 1; // The consumer group id
    Error  error       = 2; // Indicates if there was something wrong with the requested group
    string coordinator = 3; // The broker that is the group coordinator
    uint64 epoch       = 4 [jstype=JS_STRING]; // The group generation
}

// PartitionEventTimestamps contains timestamps related to a stream partition.
message PartitionEventTimestamps {
    int64 firstTimestamp  = 1 [jstype=JS_STRING]; // When the first event has occured
    int64 latestTimestamp = 2 [jstype=JS_STRING]; // When the latest event has occured
}

// PartitionMetadata contains information for a stream partition.
message PartitionMetadata {
    int32                    id                         = 1; // Partition id
    string                   leader                     = 2; // Broker id of the partition leader
    repeated string          replicas                   = 3; // Broker ids of the partition replicas
    repeated string          isr                        = 4; // Broker ids of the in-sync replica set
    int64                    highWatermark              = 5 [jstype=JS_STRING]; // High watermark of the partition
    int64                    newestOffset               = 6 [jstype=JS_STRING]; // Newest offset of the partition
    bool                     paused                     = 7; // Is this partition paused?
    bool                     readonly                   = 8; // Is this partition read-only?
    PartitionEventTimestamps messagesReceivedTimestamps = 9; // Timestamps of when the first and latest message have been received
    PartitionEventTimestamps pauseTimestamps            = 10; // Timestamps of when the paused status has been changed
    PartitionEventTimestamps readonlyTimestamps         = 11; // Timestamps of when the read-only status has been changed
}

// AckPolicy controls the behavior of message acknowledgements.
enum AckPolicy {
    LEADER = 0; // The ack will be sent once the leader has written the message to its log
    ALL    = 1; // The ack will be sent after the ISR replicas have written the message to their logs
    NONE   = 2; // No ack will be sent
}

// Message represents a message from a stream.
message Message {
    int64               offset        = 1 [jstype=JS_STRING];  // Monotonic message offset in the stream
    bytes               key           = 2;  // Message key
    bytes               value         = 3;  // Message payload
    int64               timestamp     = 4 [jstype=JS_STRING];  // When the message was received by the broker
    string              stream        = 5;  // Stream name message was received on
    int32               partition     = 6;  // Stream partition message was assigned to
    string              subject       = 7;  // NATS subject message was received on
    string              replySubject  = 8;  // NATS reply subject
    map<string, bytes>  headers       = 9;  // Message headers
    string              ackInbox      = 10; // NATS subject to publish acks to
    string              correlationId = 11; // User-supplied value to correlate acks to publishes
    AckPolicy           ackPolicy     = 12; // Controls the behavior of acks
}

// Ack represents an acknowledgement that a message was committed to a stream
// partition.
message Ack {
    // Error represents an error that is piggybacked to AckInbox due to any
    // error that is encountered upon publishing, e.g. a PublishRequest may
    // include the expected offset which is outdated on the partition. This
    // publish would be refused.
    enum Error {
        OK               = 0;
        UNKNOWN          = 1;
        INCORRECT_OFFSET = 2;
        TOO_LARGE        = 3;
        ENCRYPTION       = 4;
    }

    string    stream             = 1; // Name of the stream
    string    partitionSubject   = 2; // NATS subject partition is attached to
    string    msgSubject         = 3; // NATS subject the message was received on
    int64     offset             = 4 [jstype=JS_STRING]; // Stream offset the message was committed to
    string    ackInbox           = 5; // NATS subject to publish acks to
    string    correlationId      = 6; // User-supplied value from the message
    AckPolicy ackPolicy          = 7; // The AckPolicy sent on the message
    int64     receptionTimestamp = 8 [jstype=JS_STRING]; // When the message was received by the server
    int64     commitTimestamp    = 9 [jstype=JS_STRING]; // When the message was committed
    Error     ackError           = 10; // Error sent in case the partition refused to accept the published message
}

// ActivityStreamOp represents an activity stream operation.
enum ActivityStreamOp {
    CREATE_STREAM        = 0;
    DELETE_STREAM        = 1;
    PAUSE_STREAM         = 2;
    RESUME_STREAM        = 3;
    SET_STREAM_READONLY  = 4;
    JOIN_CONSUMER_GROUP  = 5;
    LEAVE_CONSUMER_GROUP = 6;
}

// CreateStreamOp represents a stream creation operation.
message CreateStreamOp {
    string          stream     = 1;
    repeated int32  partitions = 2;
}

// DeleteStreamOp represents a stream deletion operation.
message DeleteStreamOp {
    string stream = 1;
}

// PauseStreamOp represents a stream pause operation.
message PauseStreamOp {
    string         stream     = 1;
    repeated int32 partitions = 2;
    bool           resumeAll  = 3;
}

// ResumeStreamOp represents a stream resume operation.
message ResumeStreamOp {
    string          stream     = 1;
    repeated int32  partitions = 2;
}

// ReadonlyStreamOp represents a stream readonly operation.
message SetStreamReadonlyOp {
    string         stream     = 1;
    repeated int32 partitions = 2;
    bool           readonly   = 3;
}

// JoinConsumerGroupOp represents a consumer group join operation.
message JoinConsumerGroupOp {
    string          groupId    = 1;
    string          consumerId = 2;
    repeated string streams    = 3;
}

// LeaveConsumerGroupOp represents a consumer group leave operation.
message LeaveConsumerGroupOp {
    string groupId    = 1;
    string consumerId = 2;
    bool   expired    = 3; // If consumer was removed because they timed out.
}

// ActivityStreamEvent represents an entry into the activity stream.
message ActivityStreamEvent {
    uint64               id                   = 1 [jstype=JS_STRING];
    ActivityStreamOp     op                   = 2;
    CreateStreamOp       createStreamOp       = 3;
    DeleteStreamOp       deleteStreamOp       = 4;
    PauseStreamOp        pauseStreamOp        = 5;
    ResumeStreamOp       resumeStreamOp       = 6;
    SetStreamReadonlyOp  setStreamReadonlyOp  = 7;
    JoinConsumerGroupOp  joinConsumerGroupOp  = 8;
    LeaveConsumerGroupOp leaveConsumerGroupOp = 9;
}

// API is the main Liftbridge server interface clients interact with.
service API {
    // CreateStream creates a new stream attached to a NATS subject. It returns
    // an AlreadyExists status code if a stream with the given subject and name
    // already exists.
    rpc CreateStream(CreateStreamRequest) returns (CreateStreamResponse) {}

    // DeleteStream deletes a stream.
    rpc DeleteStream(DeleteStreamRequest) returns (DeleteStreamResponse) {}

    // PauseStream pauses a stream's partitions. If no partitions are
    // specified, all of the stream's partitions will be paused. Partitions are
    // resumed when they are published to via the Liftbridge Publish API.
    rpc PauseStream(PauseStreamRequest) returns (PauseStreamResponse) {}

    // SetStreamReadonly sets a read-only flag to a partition. Returns a
    // NoSuchStream error code if the given stream or partition does not exist.
    rpc SetStreamReadonly(SetStreamReadonlyRequest) returns (SetStreamReadonlyResponse) {}

    // Subscribe creates an ephemeral subscription for the given stream. It
    // begins to receive messages starting at the given offset and waits for
    // new messages when it reaches the end of the stream. Use the request
    // context to close the subscription.
    rpc Subscribe(SubscribeRequest) returns (stream Message) {}

    // FetchMetadata retrieves the latest cluster metadata, including stream
    // broker information.
    rpc FetchMetadata(FetchMetadataRequest) returns (FetchMetadataResponse) {}

    // FetchPartitionMetadata retrieves the latest partition metadata from partition leader
    // The main interest is to retrieve Highest Watermark and Newest Offset
    rpc FetchPartitionMetadata(FetchPartitionMetadataRequest) returns (FetchPartitionMetadataResponse) {}

    // Publish a new message to a stream. If the AckPolicy is not NONE and a
    // deadline is provided, this will synchronously block until the ack is
    // received. If the ack is not received in time, a DeadlineExceeded status
    // code is returned.
    rpc Publish(PublishRequest) returns (PublishResponse) {}

    // PublishAsync is used to asynchronously publish messages to a stream in a
    // pipelined fashion. This returns a stream which will yield
    // PublishResponses for messages whose AckPolicy is not NONE.
    rpc PublishAsync(stream PublishRequest) returns (stream PublishResponse) {}

    // Publish a Liftbridge message to a NATS subject. If the AckPolicy is not NONE and a
    // deadline is provided, this will synchronously block until the first ack
    // is received. If an ack is not received in time, a DeadlineExceeded
    // status code is returned.
    rpc PublishToSubject(PublishToSubjectRequest) returns (PublishToSubjectResponse) {}

    // SetCursor stores a cursor position for a particular stream partition
    // which is uniquely identified by an opaque string.
    //
    // NOTE: This is a beta endpoint and is subject to change. It is not
    // included as part of Liftbridge's semantic versioning scheme.
    rpc SetCursor(SetCursorRequest) returns (SetCursorResponse) {}

    // FetchCursor retrieves a partition cursor position.
    //
    // NOTE: This is a beta endpoint and is subject to change. It is not
    // included as part of Liftbridge's semantic versioning scheme.
    rpc FetchCursor(FetchCursorRequest) returns (FetchCursorResponse) {}

    // JoinConsumerGroup adds a consumer to a consumer group. If the group does
    // not exist, it will create it first.
    //
    // NOTE: This is a beta endpoint and is subject to change. It is not
    // included as part of Liftbridge's semantic versioning scheme.
    rpc JoinConsumerGroup(JoinConsumerGroupRequest) returns (JoinConsumerGroupResponse) {}

    // LeaveConsumerGroup removes a consumer from a consumer group.
    //
    // NOTE: This is a beta endpoint and is subject to change. It is not
    // included as part of Liftbridge's semantic versioning scheme.
    rpc LeaveConsumerGroup(LeaveConsumerGroupRequest) returns (LeaveConsumerGroupResponse) {}

    // FetchConsumerGroupAssignments retrieves the partition assignments for a
    // consumer. This also acts as a heartbeat for the consumer so that the
    // coordinator keeps the consumer active in the group.
    //
    // NOTE: This is a beta endpoint and is subject to change. It is not
    // included as part of Liftbridge's semantic versioning scheme.
    rpc FetchConsumerGroupAssignments(FetchConsumerGroupAssignmentsRequest) returns (FetchConsumerGroupAssignmentsResponse) {}

    // ReportConsumerGroupCoordinator reports a consumer group coordinator as
    // failed. If a majority of the group's members report the coordinator
    // within a bounded period, the cluster will select a new coordinator.
    //
    // NOTE: This is a beta endpoint and is subject to change. It is not
    // included as part of Liftbridge's semantic versioning scheme.
    rpc ReportConsumerGroupCoordinator(ReportConsumerGroupCoordinatorRequest) returns (ReportConsumerGroupCoordinatorResponse) {}
}