syntax = "proto3";

package kafkalite;

message Header {
  string key = 1;
  bytes value = 2;
}

message NewMessage {
  string topic = 1;
  string key = 2;
  bytes payload_json = 3;
  repeated Header headers = 4;
  int64 published_at_unix_ms = 5;
}

message AppendAck {
  string topic = 1;
  int32 partition = 2;
  uint64 offset = 3;
  int64 published_at_unix_ms = 4;
}

message StoredMessage {
  string topic = 1;
  int32 partition = 2;
  uint64 offset = 3;
  string key = 4;
  bytes payload_json = 5;
  repeated Header headers = 6;
  int64 published_at_unix_ms = 7;
}

message Assignment {
  string topic = 1;
  int32 partition = 2;
  uint64 next_offset = 3;
}

message JoinGroupRequest {
  string group_id = 1;
  string member_id = 2;
  repeated string topics = 3;
  int32 partition = 4;
  uint32 session_timeout_ms = 5;
  int64 joined_at_unix_ms = 6;
}

message JoinGroupResponse {
  uint64 generation = 1;
  repeated Assignment assignments = 2;
}

message HeartbeatRequest {
  string group_id = 1;
  string member_id = 2;
  uint64 generation = 3;
  int64 heartbeat_at_unix_ms = 4;
}

message HeartbeatResponse {
  uint64 generation = 1;
  bool needs_rejoin = 2;
}

message PollMessagesRequest {
  string group_id = 1;
  string member_id = 2;
  uint64 generation = 3;
  string topic = 4;
  int32 partition = 5;
  uint64 start_offset = 6;
  uint32 limit = 7;
  uint32 wait_ms = 8;
}

message PollMessagesResponse {
  repeated StoredMessage messages = 1;
}

message CommitOffsetsRequest {
  string group_id = 1;
  string member_id = 2;
  uint64 generation = 3;
  string topic = 4;
  int32 partition = 5;
  uint64 next_offset = 6;
  int64 updated_at_unix_ms = 7;
}

message CommitOffsetsResponse {
  bool committed = 1;
  string topic = 2;
  int32 partition = 3;
  uint64 next_offset = 4;
}

message LeaveGroupRequest {
  string group_id = 1;
  string member_id = 2;
}

message LeaveGroupResponse {
  bool left = 1;
}

message GetCommittedOffsetRequest {
  string group_id = 1;
  string topic = 2;
  int32 partition = 3;
}

message GetCommittedOffsetResponse {
  bool found = 1;
  string topic = 2;
  int32 partition = 3;
  optional uint64 next_offset = 4;
}

message ReadMessagesRequest {
  string topic = 1;
  uint64 start_offset = 2;
  uint32 limit = 3;
}

message ReadMessagesResponse {
  repeated StoredMessage messages = 1;
}

message HealthCheckRequest {}

message HealthCheckResponse {
  string status = 1;
  string storage_backend = 2;
  string database_path = 3;
}

service Kafkalite {
  rpc Health(HealthCheckRequest) returns (HealthCheckResponse);
  rpc Publish(NewMessage) returns (AppendAck);
  rpc JoinGroup(JoinGroupRequest) returns (JoinGroupResponse);
  rpc Heartbeat(HeartbeatRequest) returns (HeartbeatResponse);
  rpc PollMessages(PollMessagesRequest) returns (PollMessagesResponse);
  rpc CommitOffsets(CommitOffsetsRequest) returns (CommitOffsetsResponse);
  rpc LeaveGroup(LeaveGroupRequest) returns (LeaveGroupResponse);
  rpc GetCommittedOffset(GetCommittedOffsetRequest) returns (GetCommittedOffsetResponse);
}
