feat: add tls and api key authentication support

README.md

@@ -44,6 +44,77 @@ try (FilaClient client = FilaClient.builder("localhost:5555").build()) {
 }
 ```
 
+## TLS
+
+### System trust store (public CAs)
+
+If the Fila server uses a certificate issued by a public CA (e.g., Let's Encrypt), enable TLS with the JVM's default trust store:
+
+```java
+try (FilaClient client = FilaClient.builder("localhost:5555")
+    .withTls()
+    .build()) {
+    // use client...
+}
+```
+
+### Custom CA certificate
+
+For servers using self-signed or private CA certificates, provide the CA cert explicitly:
+
+```java
+byte[] caCert = Files.readAllBytes(Path.of("ca.pem"));
+
+try (FilaClient client = FilaClient.builder("localhost:5555")
+    .withTlsCaCert(caCert)
+    .build()) {
+    // use client...
+}
+```
+
+### Mutual TLS (mTLS)
+
+For mutual TLS, also provide the client certificate and key. This works with both trust modes:
+
+```java
+byte[] caCert = Files.readAllBytes(Path.of("ca.pem"));
+byte[] clientCert = Files.readAllBytes(Path.of("client.pem"));
+byte[] clientKey = Files.readAllBytes(Path.of("client-key.pem"));
+
+try (FilaClient client = FilaClient.builder("localhost:5555")
+    .withTlsCaCert(caCert)
+    .withTlsClientCert(clientCert, clientKey)
+    .build()) {
+    // use client...
+}
+```
+
+## API Key Authentication
+
+When the server has auth enabled, provide an API key:
+
+```java
+try (FilaClient client = FilaClient.builder("localhost:5555")
+    .withApiKey("your-api-key")
+    .build()) {
+    // use client...
+}
+```
+
+The key is sent as a `Bearer` token in the `authorization` metadata header on every RPC.
+
+TLS and API key auth can be combined:
+
+```java
+try (FilaClient client = FilaClient.builder("localhost:5555")
+    .withTlsCaCert(caCert)
+    .withTlsClientCert(clientCert, clientKey)
+    .withApiKey("your-api-key")
+    .build()) {
+    // use client...
+}
+```
+
 ## API Reference
 
 ### `FilaClient`
@@ -56,6 +127,17 @@ FilaClient client = FilaClient.builder("localhost:5555").build();
 
 `FilaClient` implements `AutoCloseable` for use with try-with-resources.
 
+#### Builder Methods
+
+| Method | Description |
+|--------|-------------|
+| `withTls()` | Enable TLS using JVM's default trust store (cacerts) |
+| `withTlsCaCert(byte[] caCertPem)` | CA certificate for TLS server verification (implies `withTls()`) |
+| `withTlsClientCert(byte[] certPem, byte[] keyPem)` | Client cert + key for mTLS |
+| `withApiKey(String apiKey)` | API key sent as `Bearer` token on every RPC |
+
+All builder methods are optional. When none are set, the client connects over plaintext without authentication (backward compatible).
+
 #### `enqueue(String queue, Map<String, String> headers, byte[] payload) -> String`
 
 Enqueue a message. Returns the broker-assigned message ID (UUIDv7).

proto/fila/v1/admin.proto

@@ -11,6 +11,15 @@ service FilaAdmin {
   rpc GetStats(GetStatsRequest) returns (GetStatsResponse);
   rpc Redrive(RedriveRequest) returns (RedriveResponse);
   rpc ListQueues(ListQueuesRequest) returns (ListQueuesResponse);
+
+  // API key management. CreateApiKey bypasses auth (bootstrap); others require a valid key.
+  rpc CreateApiKey(CreateApiKeyRequest) returns (CreateApiKeyResponse);
+  rpc RevokeApiKey(RevokeApiKeyRequest) returns (RevokeApiKeyResponse);
+  rpc ListApiKeys(ListApiKeysRequest) returns (ListApiKeysResponse);
+
+  // Per-key ACL management.
+  rpc SetAcl(SetAclRequest) returns (SetAclResponse);
+  rpc GetAcl(GetAclRequest) returns (GetAclResponse);
 }
 
 message CreateQueueRequest {
@@ -89,6 +98,9 @@ message GetStatsResponse {
   uint32 quantum = 5;
   repeated PerFairnessKeyStats per_key_stats = 6;
   repeated PerThrottleKeyStats per_throttle_stats = 7;
+  // Cluster fields (0 when not in cluster mode).
+  uint64 leader_node_id = 8;
+  uint32 replication_count = 9;
 }
 
 message RedriveRequest {
@@ -107,8 +119,79 @@ message QueueInfo {
   uint64 depth = 2;
   uint64 in_flight = 3;
   uint32 active_consumers = 4;
+  uint64 leader_node_id = 5;
 }
 
 message ListQueuesResponse {
   repeated QueueInfo queues = 1;
+  uint32 cluster_node_count = 2;
+}
+
+// --- API Key Management ---
+
+message CreateApiKeyRequest {
+  /// Human-readable label for the key.
+  string name = 1;
+  /// Optional Unix timestamp (milliseconds) after which the key expires.
+  /// 0 means no expiration.
+  uint64 expires_at_ms = 2;
+  /// When true, the key bypasses all ACL checks (superadmin).
+  bool is_superadmin = 3;
+}
+
+message CreateApiKeyResponse {
+  /// Opaque key ID for management operations (revoke, list, set-acl).
+  string key_id = 1;
+  /// Plaintext API key. Returned once — store it securely.
+  string key = 2;
+  /// Whether this key has superadmin privileges.
+  bool is_superadmin = 3;
+}
+
+message RevokeApiKeyRequest {
+  string key_id = 1;
+}
+
+message RevokeApiKeyResponse {}
+
+message ListApiKeysRequest {}
+
+message ApiKeyInfo {
+  string key_id = 1;
+  string name = 2;
+  uint64 created_at_ms = 3;
+  /// 0 means no expiration.
+  uint64 expires_at_ms = 4;
+  bool is_superadmin = 5;
+}
+
+message ListApiKeysResponse {
+  repeated ApiKeyInfo keys = 1;
+}
+
+// --- ACL Management ---
+
+/// A single permission grant: kind (produce/consume/admin) + queue pattern.
+message AclPermission {
+  /// One of: "produce", "consume", "admin".
+  string kind = 1;
+  /// Queue name or wildcard ("*" or "orders.*").
+  string pattern = 2;
+}
+
+message SetAclRequest {
+  string key_id = 1;
+  repeated AclPermission permissions = 2;
+}
+
+message SetAclResponse {}
+
+message GetAclRequest {
+  string key_id = 1;
+}
+
+message GetAclResponse {
+  string key_id = 1;
+  repeated AclPermission permissions = 2;
+  bool is_superadmin = 3;
 }

src/main/java/dev/faisca/fila/ApiKeyInterceptor.java

@@ -0,0 +1,36 @@
+package dev.faisca.fila;
+
+import io.grpc.CallOptions;
+import io.grpc.Channel;
+import io.grpc.ClientCall;
+import io.grpc.ClientInterceptor;
+import io.grpc.ForwardingClientCall.SimpleForwardingClientCall;
+import io.grpc.Metadata;
+import io.grpc.MethodDescriptor;
+
+/**
+ * gRPC client interceptor that attaches a {@code Bearer} API key to the {@code authorization}
+ * metadata header on every outgoing RPC.
+ */
+final class ApiKeyInterceptor implements ClientInterceptor {
+  private static final Metadata.Key<String> AUTH_KEY =
+      Metadata.Key.of("authorization", Metadata.ASCII_STRING_MARSHALLER);
+
+  private final String headerValue;
+
+  ApiKeyInterceptor(String apiKey) {
+    this.headerValue = "Bearer " + apiKey;
+  }
+
+  @Override
+  public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
+      MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
+    return new SimpleForwardingClientCall<>(next.newCall(method, callOptions)) {
+      @Override
+      public void start(Listener<RespT> responseListener, Metadata headers) {
+        headers.put(AUTH_KEY, headerValue);
+        super.start(responseListener, headers);
+      }
+    };
+  }
+}