You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/07/02 22:08:48 UTC
(iggy) 41/43: feat(server-ng): list personal access tokens over HTTP and TCP
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch http-listener-rebased
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 1b9874fdb3c23c0ae6e92e52ad2f6efc8197a56f
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Thu Jul 2 22:43:05 2026 +0200
feat(server-ng): list personal access tokens over HTTP and TCP
GET_PERSONAL_ACCESS_TOKENS was the one read the ng server did not
serve: native TCP fell through to an empty reply (a false success)
and the HTTP route was parked. PATs sit in AHashMaps guarded by the
deterministic-apply invariant, so the users STM gains a read-only,
name-sorted accessor instead of exposing raw map iteration order.
The list is caller-scoped, so the shared read builder now carries
the authenticated user id; the transport gates require only
authentication, matching the legacy handler (no permissioner rule).
---
core/integration/tests/server/http_rbac.rs | 92 ++++++++++++++++++++++++++++++
core/metadata/src/stm/user.rs | 64 +++++++++++++++++++++
core/server-ng/src/dispatch.rs | 15 +++--
core/server-ng/src/http.rs | 54 +++++++++++++-----
core/server-ng/src/responses.rs | 91 ++++++++++++++++++++++++++---
5 files changed, 291 insertions(+), 25 deletions(-)
diff --git a/core/integration/tests/server/http_rbac.rs b/core/integration/tests/server/http_rbac.rs
index f408bc5e7..09ff99351 100644
--- a/core/integration/tests/server/http_rbac.rs
+++ b/core/integration/tests/server/http_rbac.rs
@@ -946,6 +946,98 @@ async fn given_non_root_tcp_session_when_polling_denied_should_surface_typed_una
);
}
+// === Case family 9: PAT list is caller-scoped ================================
+
+/// Listing is self-scoped like the other PAT ops: any authenticated user sees
+/// exactly their own tokens (name + expiry, name-sorted), never another
+/// user's. Asserted over HTTP and via the SDK on TCP, which share the read
+/// builder.
+#[iggy_harness]
+async fn given_two_users_with_tokens_when_listing_should_return_only_callers(
+ harness: &TestHarness,
+) {
+ let root = Client::login_root(harness).await;
+ root.create_user("alice", "alice-pass", no_permissions())
+ .await;
+ root.create_user("bob", "bob-pass", no_permissions()).await;
+ let alice = root.login("alice", "alice-pass").await;
+ let bob = root.login("bob", "bob-pass").await;
+
+ // Created in reverse name order to prove the listing sorts by name.
+ for name in ["alice-zeta", "alice-alpha"] {
+ let created = alice
+ .post_json(
+ "/personal-access-tokens",
+ &json!({ "name": name, "expiry": 0 }),
+ )
+ .await;
+ assert_eq!(
+ created.status(),
+ StatusCode::OK,
+ "alice create PAT {name} must be 200"
+ );
+ }
+ let created = bob
+ .post_json(
+ "/personal-access-tokens",
+ &json!({ "name": "bob-pat", "expiry": 0 }),
+ )
+ .await;
+ assert_eq!(
+ created.status(),
+ StatusCode::OK,
+ "bob create PAT must be 200"
+ );
+
+ assert_eq!(
+ pat_names(alice.get("/personal-access-tokens").await).await,
+ ["alice-alpha", "alice-zeta"],
+ "alice must list exactly her own tokens, name-sorted"
+ );
+ assert_eq!(
+ pat_names(bob.get("/personal-access-tokens").await).await,
+ ["bob-pat"],
+ "bob must never see alice's tokens"
+ );
+
+ // Same caller-scoped view over native TCP through the SDK.
+ let client = harness.tcp_new_client().await.expect("tcp client");
+ client
+ .login_user("alice", "alice-pass")
+ .await
+ .expect("alice tcp login");
+ let tokens = client
+ .get_personal_access_tokens()
+ .await
+ .expect("tcp PAT list");
+ let tcp_names: Vec<&str> = tokens.iter().map(|token| token.name.as_str()).collect();
+ assert_eq!(
+ tcp_names,
+ ["alice-alpha", "alice-zeta"],
+ "TCP list must match the HTTP caller-scoped view"
+ );
+}
+
+/// Decode a PAT-list response body into its token names, asserting 200 first.
+async fn pat_names(response: Response) -> Vec<String> {
+ assert_eq!(
+ response.status(),
+ StatusCode::OK,
+ "PAT list must be 200 for an authenticated caller"
+ );
+ let body: Value = response.json().await.expect("PAT list body must be JSON");
+ body.as_array()
+ .expect("PAT list must be a JSON array")
+ .iter()
+ .map(|token| {
+ token["name"]
+ .as_str()
+ .expect("every token must carry a name")
+ .to_owned()
+ })
+ .collect()
+}
+
/// The current-password check gates a self-service rotation: a wrong current
/// password is rejected pre-consensus as 400 `InvalidCredentials` (never
/// reaching the replicated apply), and only the correct one rotates the
diff --git a/core/metadata/src/stm/user.rs b/core/metadata/src/stm/user.rs
index dafd4e812..7a5ee16eb 100644
--- a/core/metadata/src/stm/user.rs
+++ b/core/metadata/src/stm/user.rs
@@ -165,6 +165,34 @@ impl UsersInner {
.map(|(_, user_id, name)| (*user_id, Arc::clone(name)))
.collect()
}
+
+ /// Collect `(name, expiry_at)` for every live personal access token of
+ /// one user, sorted by name.
+ ///
+ /// Read-only accessor for the non-replicated list path (the caller lists
+ /// its own tokens). The backing per-user map is an `AHashMap`, whose
+ /// iteration order is seeded per process, so the collected list is
+ /// sorted before it is returned. Never lift this iteration into an apply
+ /// handler: apply must stay deterministic across replicas (see the
+ /// invariant on `personal_access_token_index`).
+ #[must_use]
+ pub fn personal_access_tokens_of(
+ &self,
+ user_id: UserId,
+ ) -> Vec<(Arc<str>, Option<IggyTimestamp>)> {
+ let mut tokens: Vec<_> = self
+ .personal_access_tokens
+ .get(&user_id)
+ .map(|user_tokens| {
+ user_tokens
+ .values()
+ .map(|pat| (Arc::clone(&pat.name), pat.expiry_at))
+ .collect()
+ })
+ .unwrap_or_default();
+ tokens.sort_by(|(left, _), (right, _)| left.cmp(right));
+ tokens
+ }
}
impl Users {
@@ -971,6 +999,42 @@ mod tests {
);
}
+ #[test]
+ fn personal_access_tokens_of_lists_only_callers_sorted_by_name() {
+ let mut users = UsersInner::new();
+ // Insertion order deliberately not name order; a token for user 9
+ // proves the listing is caller-scoped.
+ for (user_id, name, expiry, fill) in [
+ (5u32, "zeta", 0u64, b'a'),
+ (5u32, "alpha", 1, b'b'),
+ (9u32, "other", 0, b'c'),
+ ] {
+ CreatePersonalAccessTokenRequest {
+ user_id,
+ name: WireName::new(name).unwrap(),
+ expiry,
+ token_hash: [fill; PAT_TOKEN_HASH_BYTES],
+ }
+ .apply(&mut users, IggyTimestamp::zero());
+ }
+
+ let tokens = users.personal_access_tokens_of(5);
+ let names: Vec<&str> = tokens.iter().map(|(name, _)| name.as_ref()).collect();
+ assert_eq!(names, ["alpha", "zeta"], "own tokens only, sorted by name");
+ assert!(
+ tokens[0].1.is_some(),
+ "expiring token must carry its expiry_at"
+ );
+ assert!(
+ tokens[1].1.is_none(),
+ "never-expiring token must carry no expiry_at"
+ );
+ assert!(
+ users.personal_access_tokens_of(42).is_empty(),
+ "a user without tokens lists empty"
+ );
+ }
+
fn create_user(users: &mut UsersInner, username: &str) {
let request = CreateUserRequest {
username: WireName::new(username).unwrap(),
diff --git a/core/server-ng/src/dispatch.rs b/core/server-ng/src/dispatch.rs
index ce4e64195..b13088f29 100644
--- a/core/server-ng/src/dispatch.rs
+++ b/core/server-ng/src/dispatch.rs
@@ -49,9 +49,10 @@ use consensus::{
};
use iggy_binary_protocol::codes::{
GET_CLIENT_CODE, GET_CLIENTS_CODE, GET_CLUSTER_METADATA_CODE, GET_CONSUMER_GROUP_CODE,
- GET_CONSUMER_GROUPS_CODE, GET_CONSUMER_OFFSET_CODE, GET_ME_CODE, GET_STATS_CODE,
- GET_STREAM_CODE, GET_STREAMS_CODE, GET_TOPIC_CODE, GET_TOPICS_CODE, GET_USER_CODE,
- GET_USERS_CODE, PING_CODE, POLL_MESSAGES_CODE, SYNC_CONSUMER_GROUP_CODE,
+ GET_CONSUMER_GROUPS_CODE, GET_CONSUMER_OFFSET_CODE, GET_ME_CODE,
+ GET_PERSONAL_ACCESS_TOKENS_CODE, GET_STATS_CODE, GET_STREAM_CODE, GET_STREAMS_CODE,
+ GET_TOPIC_CODE, GET_TOPICS_CODE, GET_USER_CODE, GET_USERS_CODE, PING_CODE, POLL_MESSAGES_CODE,
+ SYNC_CONSUMER_GROUP_CODE,
};
use iggy_binary_protocol::primitives::consumer::WireConsumer;
use iggy_binary_protocol::primitives::polling_strategy::WirePollingStrategy;
@@ -971,7 +972,7 @@ async fn handle_default_non_replicated(
send_non_replicated_deny(shard, request, transport_client_id, error.as_code()).await;
return;
}
- match build_non_replicated_response(shard, code, request_body(request)) {
+ match build_non_replicated_response(shard, code, request_body(request), user_id) {
Ok(response) => {
let commit = current_metadata_commit(shard);
let reply = response.into_reply(
@@ -1519,7 +1520,8 @@ fn authorize_partition_read(
/// miss, which falls through to the builder's own not-found reply so the legacy
/// notfound-before-permission ordering holds. `Err` denies with that code.
/// Unscoped rules gate directly; identifier-scoped rules resolve (stream[,
-/// topic]) against committed state first. `GET_CLUSTER_METADATA` is auth-only
+/// topic]) against committed state first. The PAT list is self-scoped, so
+/// authentication is its whole rule. `GET_CLUSTER_METADATA` is auth-only
/// (bootstrap / leader discovery) and every other code the builder serves is
/// ungated here.
fn authorize_default_read(
@@ -1532,6 +1534,9 @@ fn authorize_default_read(
GET_STATS_CODE => authorize_uid(shard, user_id, Permissioner::get_stats),
GET_USERS_CODE => authorize_uid(shard, user_id, Permissioner::get_users),
GET_USER_CODE => authorize_uid(shard, user_id, Permissioner::get_user),
+ // Self-scoped: lists only the caller's own tokens, so there is no
+ // permissioner rule to run (legacy runs none either).
+ GET_PERSONAL_ACCESS_TOKENS_CODE => user_id.map(|_| ()).ok_or(IggyError::Unauthenticated),
GET_STREAMS_CODE => authorize_uid(shard, user_id, Permissioner::get_streams),
GET_STREAM_CODE => {
let Ok(request) = GetStreamRequest::decode_from(body) else {
diff --git a/core/server-ng/src/http.rs b/core/server-ng/src/http.rs
index 0fd2b846b..02ffc1f52 100644
--- a/core/server-ng/src/http.rs
+++ b/core/server-ng/src/http.rs
@@ -42,8 +42,9 @@ use configs::http::HttpConfig;
use consensus::{CLIENTS_TABLE_MAX, MetadataHandle, VsrConsensus};
use futures::channel::oneshot;
use iggy_binary_protocol::codes::{
- GET_CONSUMER_GROUP_CODE, GET_CONSUMER_GROUPS_CODE, GET_STATS_CODE, GET_STREAM_CODE,
- GET_STREAMS_CODE, GET_TOPIC_CODE, GET_TOPICS_CODE, GET_USER_CODE, GET_USERS_CODE,
+ GET_CONSUMER_GROUP_CODE, GET_CONSUMER_GROUPS_CODE, GET_PERSONAL_ACCESS_TOKENS_CODE,
+ GET_STATS_CODE, GET_STREAM_CODE, GET_STREAMS_CODE, GET_TOPIC_CODE, GET_TOPICS_CODE,
+ GET_USER_CODE, GET_USERS_CODE,
};
use iggy_binary_protocol::consensus::{
Command2, EvictionHeader, EvictionReason, HEADER_SIZE, result_code, result_section_len,
@@ -62,6 +63,7 @@ use iggy_binary_protocol::requests::messages::{
};
use iggy_binary_protocol::requests::personal_access_tokens::{
CreatePersonalAccessTokenRequest, DeletePersonalAccessTokenRequest,
+ GetPersonalAccessTokensRequest,
};
use iggy_binary_protocol::requests::streams::{
CreateStreamRequest, DeleteStreamRequest, GetStreamRequest, GetStreamsRequest,
@@ -81,7 +83,9 @@ use iggy_binary_protocol::responses::clients::get_client::ClientDetailsResponse;
use iggy_binary_protocol::responses::clients::get_clients::GetClientsResponse;
use iggy_binary_protocol::responses::consumer_groups::get_consumer_group::ConsumerGroupDetailsResponse;
use iggy_binary_protocol::responses::consumer_groups::get_consumer_groups::GetConsumerGroupsResponse;
-use iggy_binary_protocol::responses::personal_access_tokens::RawPersonalAccessTokenResponse;
+use iggy_binary_protocol::responses::personal_access_tokens::{
+ GetPersonalAccessTokensResponse, RawPersonalAccessTokenResponse,
+};
use iggy_binary_protocol::responses::streams::get_stream::GetStreamResponse;
use iggy_binary_protocol::responses::streams::get_streams::GetStreamsResponse;
use iggy_binary_protocol::responses::system::get_stats::StatsResponse;
@@ -114,16 +118,16 @@ use iggy_common::update_topic::UpdateTopic;
use iggy_common::update_user::UpdateUser;
use iggy_common::wire_conversions::{
clients_from_wire, consumer_groups_from_wire, consumer_to_wire, identifier_to_wire,
- partitioning_to_wire, permissions_to_wire, streams_from_wire, topics_from_wire,
- users_from_wire,
+ partitioning_to_wire, permissions_to_wire, personal_access_tokens_from_wire, streams_from_wire,
+ topics_from_wire, users_from_wire,
};
use iggy_common::{
ClientInfo, ClientInfoDetails, ClusterMetadata, ClusterNode, ClusterNodeRole,
ClusterNodeStatus, Consumer, ConsumerGroup, ConsumerGroupDetails, ConsumerOffsetInfo,
- Identifier, IdentityInfo, IggyError, IggyMessageView, IggyTimestamp, PollMessages,
- PolledMessages, RESYNC_REQUIRED_PARTITION_SENTINEL, RawPersonalAccessToken, SendMessages,
- Stats, Stream, StreamDetails, TokenInfo, Topic, TopicDetails, TransportEndpoints, UserInfo,
- UserInfoDetails, Validatable,
+ Identifier, IdentityInfo, IggyError, IggyMessageView, IggyTimestamp, PersonalAccessTokenInfo,
+ PollMessages, PolledMessages, RESYNC_REQUIRED_PARTITION_SENTINEL, RawPersonalAccessToken,
+ SendMessages, Stats, Stream, StreamDetails, TokenInfo, Topic, TopicDetails, TransportEndpoints,
+ UserInfo, UserInfoDetails, Validatable,
};
use message_bus::{BusMessage, InstanceToken, client_listener};
use metadata::impls::metadata::StreamsFrontend;
@@ -1083,9 +1087,7 @@ fn router(state: HttpState, max_request_size: usize) -> Router {
"/streams/{stream_id}/topics/{topic_id}/consumer-groups/{group_id}",
get(get_cg).delete(delete_cg),
)
- // TODO(hubcio): GET /personal-access-tokens (list) deferred - needs a server-ng STM
- // list accessor (PATs sit behind the "Never iterate" apply invariant; TCP-ng lacks it too).
- .route("/personal-access-tokens", post(create_pat))
+ .route("/personal-access-tokens", get(get_pats).post(create_pat))
.route("/personal-access-tokens/{name}", delete(delete_pat))
.route("/stats", get(get_stats))
.route("/cluster/metadata", get(get_cluster_metadata))
@@ -2193,6 +2195,30 @@ async fn update_permissions(
Ok(StatusCode::NO_CONTENT)
}
+/// `GET /personal-access-tokens`: list the caller's own tokens (name +
+/// expiry, never the secrets) as the same `Vec<PersonalAccessTokenInfo>` JSON
+/// the legacy server returns. Self-scoped, so authentication is the whole
+/// rule - no permissioner check, matching legacy parity. A consensus-free
+/// local STM read via [`read_local`].
+async fn get_pats(
+ State(state): State<HttpState>,
+ identity: Identity,
+ Query(query): Query<ConsistencyQuery>,
+) -> Result<Json<Vec<PersonalAccessTokenInfo>>, ReadError> {
+ let body = GetPersonalAccessTokensRequest.to_bytes();
+ let bytes = read_local(
+ &state,
+ &identity,
+ query.consistency,
+ GET_PERSONAL_ACCESS_TOKENS_CODE,
+ &body,
+ |_, _| Ok(()),
+ )?;
+ let response = GetPersonalAccessTokensResponse::decode_from(&bytes)
+ .map_err(|_| ReadError::Rejected(IggyError::InvalidCommand))?;
+ Ok(Json(personal_access_tokens_from_wire(response)))
+}
+
/// `POST /personal-access-tokens`: mint a personal access token for the caller
/// and return its one-time raw secret as the same `{"token": ...}` JSON the
/// legacy server returns, with HTTP 200.
@@ -2749,7 +2775,9 @@ fn read_local(
rule: impl FnOnce(&Permissioner, u32) -> Result<(), IggyError>,
) -> Result<Bytes, ReadError> {
authorize_read(state, identity, consistency, rule)?;
- match build_non_replicated_response(&state.shard, code, body).map_err(ReadError::Rejected)? {
+ match build_non_replicated_response(&state.shard, code, body, Some(identity.user_id))
+ .map_err(ReadError::Rejected)?
+ {
NonReplicatedResponse::Empty => Err(ReadError::NotFound),
NonReplicatedResponse::Bytes(bytes) => Ok(bytes),
}
diff --git a/core/server-ng/src/responses.rs b/core/server-ng/src/responses.rs
index ed7e3225a..27f014812 100644
--- a/core/server-ng/src/responses.rs
+++ b/core/server-ng/src/responses.rs
@@ -18,8 +18,8 @@
//! Wire-response builders for the non-replicated read path.
//!
//! Assemble `get_me` / `get_clients` / `get_stream(s)` / `get_topic(s)` /
-//! `get_user(s)` / stats / cluster-metadata responses from per-shard
-//! session state and the metadata state machine, plus the
+//! `get_user(s)` / `get_personal_access_tokens` / stats / cluster-metadata
+//! responses from per-shard session state and the metadata state machine, plus the
//! [`NonReplicatedResponse`] dispatch shim and the partition-namespace
//! resolvers.
@@ -29,9 +29,9 @@ use crate::wire::{transport_kind_to_wire, usize_to_u32};
use bytes::{BufMut, Bytes, BytesMut};
use consensus::{MetadataHandle, VsrConsensus};
use iggy_binary_protocol::codes::{
- GET_CLUSTER_METADATA_CODE, GET_CONSUMER_GROUP_CODE, GET_CONSUMER_GROUPS_CODE, GET_STATS_CODE,
- GET_STREAM_CODE, GET_STREAMS_CODE, GET_TOPIC_CODE, GET_TOPICS_CODE, GET_USER_CODE,
- GET_USERS_CODE,
+ GET_CLUSTER_METADATA_CODE, GET_CONSUMER_GROUP_CODE, GET_CONSUMER_GROUPS_CODE,
+ GET_PERSONAL_ACCESS_TOKENS_CODE, GET_STATS_CODE, GET_STREAM_CODE, GET_STREAMS_CODE,
+ GET_TOPIC_CODE, GET_TOPICS_CODE, GET_USER_CODE, GET_USERS_CODE,
};
use iggy_binary_protocol::consensus::{RESULT_COUNT_LEN, result_code};
use iggy_binary_protocol::primitives::consumer::WireConsumer;
@@ -43,6 +43,7 @@ use iggy_binary_protocol::requests::consumer_offsets::{
StoreConsumerOffsetRequest,
};
use iggy_binary_protocol::requests::messages::SendMessagesHeader;
+use iggy_binary_protocol::requests::personal_access_tokens::GetPersonalAccessTokensRequest;
use iggy_binary_protocol::requests::segments::DeleteSegmentsRequest;
use iggy_binary_protocol::requests::streams::{GetStreamRequest, GetStreamsRequest};
use iggy_binary_protocol::requests::topics::{GetTopicRequest, GetTopicsRequest};
@@ -51,7 +52,9 @@ use iggy_binary_protocol::responses::clients::client_response::ClientResponse;
use iggy_binary_protocol::responses::clients::client_response::ConsumerGroupInfoResponse;
use iggy_binary_protocol::responses::clients::get_client::ClientDetailsResponse;
use iggy_binary_protocol::responses::consumer_groups::GetConsumerGroupsResponse;
-use iggy_binary_protocol::responses::personal_access_tokens::RawPersonalAccessTokenResponse;
+use iggy_binary_protocol::responses::personal_access_tokens::{
+ GetPersonalAccessTokensResponse, PersonalAccessTokenResponse, RawPersonalAccessTokenResponse,
+};
use iggy_binary_protocol::responses::streams::StreamResponse;
use iggy_binary_protocol::responses::streams::get_stream::{
GetStreamResponse, TopicHeader as StreamTopicHeader,
@@ -71,7 +74,7 @@ use iggy_binary_protocol::{
Command2, GenericHeader, IGGY_PROTOCOL_VERSION, KIND_CONSUMER_GROUP, Operation, ReplyHeader,
RequestHeader, WireDecode, WireEncode, WireIdentifier, WireName, WirePartitioning,
};
-use iggy_common::IggyError;
+use iggy_common::{IggyError, IggyTimestamp};
use metadata::impls::metadata::StreamsFrontend;
use partitions::PollFragments;
use server::binary::dispatch::wire_id_to_identifier;
@@ -81,6 +84,7 @@ use server_common::sharding::IggyNamespace;
use shard::ConnectedClientInfo;
use std::cell::RefCell;
use std::rc::Rc;
+use std::sync::Arc;
/// Build the `get_me` reply for the requesting connection. Identity
/// (`user_id`, transport kind, peer address) comes from the per-shard
@@ -372,10 +376,14 @@ pub(crate) fn resolve_partition_namespace(
.ok_or(IggyError::InvalidIdentifier)
}
+/// `user_id` is the authenticated caller, used only by the identity-scoped
+/// reads (currently the PAT list); every other arm ignores it. Authorization
+/// stays with the per-transport gates that run before this builder.
pub(crate) fn build_non_replicated_response(
shard: &Rc<ServerNgShard>,
code: u32,
body: &[u8],
+ user_id: Option<u32>,
) -> Result<NonReplicatedResponse, IggyError> {
match code {
GET_CLUSTER_METADATA_CODE => Ok(NonReplicatedResponse::Bytes(
@@ -427,6 +435,17 @@ pub(crate) fn build_non_replicated_response(
})
})
}
+ GET_PERSONAL_ACCESS_TOKENS_CODE => {
+ let _ = GetPersonalAccessTokensRequest::decode_from(body)
+ .map_err(|_| IggyError::InvalidCommand)?;
+ // Caller-scoped: both transport gates reject unauthenticated
+ // callers before this builder runs, so a missing id is a gate
+ // bug; fail closed rather than serve another scope.
+ let user_id = user_id.ok_or(IggyError::Unauthenticated)?;
+ Ok(NonReplicatedResponse::Bytes(
+ build_get_personal_access_tokens_response(shard, user_id)?.to_bytes(),
+ ))
+ }
GET_CONSUMER_GROUP_CODE => {
let request = GetConsumerGroupRequest::decode_from(body)
.map_err(|_| IggyError::InvalidCommand)?;
@@ -632,6 +651,39 @@ fn build_get_user_response(
})
}
+/// The caller's own tokens, never another user's - the one identity-scoped
+/// read in this module, which is why [`build_non_replicated_response`]
+/// carries a `user_id`.
+fn build_get_personal_access_tokens_response(
+ shard: &Rc<ServerNgShard>,
+ user_id: u32,
+) -> Result<GetPersonalAccessTokensResponse, IggyError> {
+ let tokens = shard
+ .plane
+ .metadata()
+ .mux_stm
+ .users()
+ .read(|users| users.personal_access_tokens_of(user_id));
+ personal_access_tokens_response(tokens)
+}
+
+fn personal_access_tokens_response(
+ tokens: Vec<(Arc<str>, Option<IggyTimestamp>)>,
+) -> Result<GetPersonalAccessTokensResponse, IggyError> {
+ let tokens = tokens
+ .into_iter()
+ .map(|(name, expiry_at)| {
+ Ok(PersonalAccessTokenResponse {
+ name: WireName::new(name.as_ref()).map_err(|_| IggyError::InvalidFormat)?,
+ // 0 is the wire encoding for a never-expiring token, matching
+ // the legacy handler and the SDK-side decode.
+ expiry_at: expiry_at.map_or(0, |expiry_at| expiry_at.as_micros()),
+ })
+ })
+ .collect::<Result<Vec<_>, IggyError>>()?;
+ Ok(GetPersonalAccessTokensResponse { tokens })
+}
+
fn build_get_topic_response(
shard: &Rc<ServerNgShard>,
stream_id: &WireIdentifier,
@@ -1182,4 +1234,29 @@ mod tests {
build_raw_pat_reply(&pat_request_header(), success, None).expect("pass-through");
assert_eq!(reply.as_slice(), original.as_slice());
}
+
+ #[test]
+ fn personal_access_tokens_response_preserves_order_and_encodes_never_as_zero() {
+ let expiry = IggyTimestamp::from(123_456u64);
+ let tokens: Vec<(Arc<str>, Option<IggyTimestamp>)> = vec![
+ (Arc::from("alpha"), Some(expiry)),
+ (Arc::from("zeta"), None),
+ ];
+
+ let response = personal_access_tokens_response(tokens).expect("mapping succeeds");
+
+ assert_eq!(response.tokens.len(), 2);
+ assert_eq!(response.tokens[0].name.as_str(), "alpha");
+ assert_eq!(response.tokens[0].expiry_at, expiry.as_micros());
+ assert_eq!(response.tokens[1].name.as_str(), "zeta");
+ assert_eq!(response.tokens[1].expiry_at, 0);
+ }
+
+ #[test]
+ fn personal_access_tokens_response_encodes_empty_list_as_empty_body() {
+ let response = personal_access_tokens_response(Vec::new()).expect("mapping succeeds");
+ // An empty body is the wire shape the SDK decodes as "no tokens"; it
+ // must stay `Bytes` (not the not-found `Empty` variant) end to end.
+ assert!(response.to_bytes().is_empty());
+ }
}