pcs_external/
client.rs

1//! [`PcsExternalClient`] — phantom-typed gRPC client for PCS External API.
2
3use std::marker::PhantomData;
4use std::sync::Arc;
5
6use prost_types::value::Kind as PbValueKind;
7use prost_types::{Struct as PbStruct, Value as PbValue};
8use ppoppo_sdk_core::grpc::{Error, PathPrefixChannel, classify_status};
9use ppoppo_sdk_core::interceptor::{AuthInterceptor, BearerCredential};
10use ppoppo_sdk_core::token_cache::TokenCache;
11
12use crate::proto::{
13    CreateSendRequestReq, GetSendRequestStatusReq, PollConfig as PbPollConfig, RecipientInput,
14    SendRequestEvent as PbSendRequestEvent, SendRequestEventType as PbEventType,
15    SendRequestStatus as PbSendStatus, StreamSendRequestEventsReq,
16    external_message_service_client::ExternalMessageServiceClient,
17};
18use crate::scopes::{GetSendStatusCapable, PcsExternalScopeSet, SendAlertCapable};
19use crate::types::{
20    DeliveryEvent, DeliveryEventKind, DeliveryStream, PollConfig, RecipientList, SendOutcome,
21    SendRequestId, SendRequestState, SendStatus, SendStatusTotals, TemplateId,
22};
23
24/// Phantom-typed gRPC client for the PCS External Developer Platform.
25///
26/// The type parameter `S: PcsExternalScopeSet` constrains which methods
27/// are callable at compile time. Construct via [`crate::PcsExternalClientBuilder`].
28///
29/// ## Capability gating
30///
31/// | Method | Required capability |
32/// |---|---|
33/// | `send_alert` | `S: SendAlertCapable` |
34/// | `get_send_status` | `S: GetSendStatusCapable` |
35///
36/// Calling a method your scope set doesn't satisfy is a compile-time
37/// E0277 — see [`crate::scopes`] for the compile_fail doctest.
38pub struct PcsExternalClient<S: PcsExternalScopeSet> {
39    pub(crate) channel: PathPrefixChannel,
40    pub(crate) cache: Arc<TokenCache>,
41    pub(crate) _scope: PhantomData<S>,
42}
43
44// Manual Clone: PathPrefixChannel and Arc<TokenCache> are both Clone; PhantomData is always Clone.
45impl<S: PcsExternalScopeSet> Clone for PcsExternalClient<S> {
46    fn clone(&self) -> Self {
47        Self {
48            channel: self.channel.clone(),
49            cache: Arc::clone(&self.cache),
50            _scope: PhantomData,
51        }
52    }
53}
54
55impl<S: SendAlertCapable> PcsExternalClient<S> {
56    /// Send a batch of templated messages to `recipients`.
57    ///
58    /// Translates to `ExternalMessageService::CreateSendRequest`.
59    ///
60    /// # Errors
61    ///
62    /// - [`Error::TokenRefresh`] — JWT acquisition failed before the call.
63    /// - [`Error::Rejected`] — PCS rejected the request (bad template, bad arg, etc.).
64    /// - [`Error::RateLimited`] — PCS rate-limit substrate denied the call;
65    ///   carries `retry_after` hint for precise retry scheduling.
66    /// - [`Error::ServerError`] — PCS 5xx, retry-eligible.
67    /// - [`Error::Transport`] — network-level failure.
68    pub async fn send_alert(
69        &self,
70        template: &TemplateId,
71        recipients: &RecipientList,
72        poll: Option<&PollConfig>,
73    ) -> Result<SendOutcome, Error> {
74        let token = self.cache.get().await?;
75        let interceptor = AuthInterceptor::new(BearerCredential::new(token));
76        let mut client =
77            ExternalMessageServiceClient::with_interceptor(self.channel.clone(), interceptor);
78
79        let req = CreateSendRequestReq {
80            template_id: template.0.clone(),
81            recipients: recipients
82                .iter()
83                .map(|r| RecipientInput {
84                    ppnum: r.ppnum.as_str().to_string(),
85                    vars: Some(vars_to_pb_struct(&r.vars)),
86                })
87                .collect(),
88            poll_config: poll.map(|p| PbPollConfig {
89                expires_in_hours: p.expires_in_hours,
90                allow_multiple: p.allow_multiple,
91            }),
92        };
93
94        let resp = client
95            .create_send_request(tonic::Request::new(req))
96            .await
97            .map_err(|s| classify_status(&s))?
98            .into_inner();
99
100        Ok(SendOutcome {
101            id: SendRequestId(resp.id),
102            state: pb_status_to_state(resp.status),
103            total_recipients: i32_to_u32(resp.total_recipients),
104        })
105    }
106}
107
108impl<S: GetSendStatusCapable> PcsExternalClient<S> {
109    /// Poll aggregate delivery status for a previously-issued send request.
110    ///
111    /// Translates to `ExternalMessageService::GetSendRequestStatus`.
112    ///
113    /// # Errors
114    ///
115    /// Same shape as [`PcsExternalClient::send_alert`].
116    pub async fn get_send_status(&self, id: &SendRequestId) -> Result<SendStatus, Error> {
117        let token = self.cache.get().await?;
118        let interceptor = AuthInterceptor::new(BearerCredential::new(token));
119        let mut client =
120            ExternalMessageServiceClient::with_interceptor(self.channel.clone(), interceptor);
121
122        let req = GetSendRequestStatusReq { id: id.0.clone() };
123        let resp = client
124            .get_send_request_status(tonic::Request::new(req))
125            .await
126            .map_err(|s| classify_status(&s))?
127            .into_inner();
128
129        let summary = resp
130            .request
131            .ok_or_else(|| Error::ProtoMismatch("GetSendRequestStatusResp.request was None".into()))?;
132
133        Ok(SendStatus {
134            id: SendRequestId(summary.id),
135            state: pb_status_to_state(summary.status),
136            totals: SendStatusTotals {
137                total: i32_to_u32(summary.total_recipients),
138                delivered: i32_to_u32(summary.delivered),
139                pending_consent: i32_to_u32(summary.pending_consent),
140                failed: i32_to_u32(summary.failed),
141            },
142        })
143    }
144
145    /// Open a server-streaming delivery event channel.
146    ///
147    /// Streams per-recipient lifecycle events for all send requests issued by
148    /// this app. Pass `after_event_id` to resume from a cursor after a
149    /// reconnect so no events are replayed.
150    ///
151    /// # Errors
152    ///
153    /// Returns [`crate::Error::TokenRefresh`] if the JWT cannot be acquired,
154    /// [`crate::Error::Rejected`] / [`crate::Error::RateLimited`] /
155    /// [`crate::Error::Transport`] if the initial RPC handshake fails.
156    /// Stream-level errors surface through [`DeliveryStream::message`].
157    pub async fn stream_send_request_events(
158        &self,
159        send_request_id: Option<&SendRequestId>,
160        after_event_id: Option<&str>,
161    ) -> Result<DeliveryStream, crate::Error> {
162        let token = self.cache.get().await?;
163        let interceptor = AuthInterceptor::new(BearerCredential::new(token));
164        let mut client =
165            ExternalMessageServiceClient::with_interceptor(self.channel.clone(), interceptor);
166
167        let req = StreamSendRequestEventsReq {
168            send_request_id: send_request_id.map(|id| id.0.clone()),
169            after_event_id: after_event_id.map(String::from),
170        };
171
172        let stream = client
173            .stream_send_request_events(tonic::Request::new(req))
174            .await
175            .map_err(|s| classify_status(&s))?
176            .into_inner();
177
178        use futures_util::StreamExt as _;
179        let mapped = stream
180            .map(|item| item.map(proto_event_to_delivery).map_err(|s| classify_status(&s)));
181
182        Ok(DeliveryStream::new(mapped))
183    }
184}
185
186fn proto_event_to_delivery(evt: PbSendRequestEvent) -> DeliveryEvent {
187    // Prost strips the enum name prefix: SEND_REQUEST_EVENT_TYPE_ → Unspecified.
188    let kind = match PbEventType::try_from(evt.event_type).unwrap_or(PbEventType::Unspecified) {
189        PbEventType::RecipientDelivered => DeliveryEventKind::RecipientDelivered {
190            ppnum: evt.recipient.as_ref().map(|r| r.ppnum.clone()).unwrap_or_default(),
191            message_id: evt.recipient.as_ref().and_then(|r| r.message_id.clone()),
192        },
193        PbEventType::RecipientFailed => DeliveryEventKind::RecipientFailed {
194            ppnum: evt.recipient.as_ref().map(|r| r.ppnum.clone()).unwrap_or_default(),
195            error_code: evt.recipient.as_ref().and_then(|r| r.error_code.clone()),
196        },
197        PbEventType::RecipientPendingConsent => DeliveryEventKind::RecipientPendingConsent {
198            ppnum: evt.recipient.map(|r| r.ppnum).unwrap_or_default(),
199        },
200        PbEventType::ConsentGranted => DeliveryEventKind::ConsentGranted,
201        PbEventType::ConsentDenied => DeliveryEventKind::ConsentDenied,
202        PbEventType::RequestCompleted => DeliveryEventKind::RequestCompleted,
203        PbEventType::PollResponseReceived => DeliveryEventKind::PollResponseReceived,
204        PbEventType::Unspecified => DeliveryEventKind::Unknown,
205    };
206    DeliveryEvent {
207        event_id: evt.event_id,
208        send_request_id: SendRequestId::new(evt.send_request_id),
209        kind,
210        occurred_at: evt.occurred_at,
211    }
212}
213
214fn vars_to_pb_struct(vars: &std::collections::BTreeMap<String, String>) -> PbStruct {
215    let fields = vars
216        .iter()
217        .map(|(k, v)| {
218            (
219                k.clone(),
220                PbValue { kind: Some(PbValueKind::StringValue(v.clone())) },
221            )
222        })
223        .collect();
224    PbStruct { fields }
225}
226
227fn pb_status_to_state(s: i32) -> SendRequestState {
228    match PbSendStatus::try_from(s).unwrap_or(PbSendStatus::Unspecified) {
229        PbSendStatus::Queued => SendRequestState::Queued,
230        PbSendStatus::Processing => SendRequestState::Processing,
231        PbSendStatus::Completed => SendRequestState::Completed,
232        PbSendStatus::Failed => SendRequestState::Failed,
233        PbSendStatus::Unspecified => SendRequestState::Unknown,
234    }
235}
236
237fn i32_to_u32(n: i32) -> u32 {
238    // PCS proto uses i32 for counters that are semantically u32. Saturate
239    // negatives to 0 — they indicate a server bug, not a real count.
240    n.max(0) as u32
241}
242
243#[cfg(test)]
244#[allow(clippy::unwrap_used, clippy::expect_used)]
245mod tests {
246    use super::*;
247    use std::assert_matches;
248
249    #[test]
250    fn pb_status_mapping_covers_all_variants() {
251        assert_eq!(pb_status_to_state(0), SendRequestState::Unknown);
252        assert_eq!(pb_status_to_state(1), SendRequestState::Queued);
253        assert_eq!(pb_status_to_state(2), SendRequestState::Processing);
254        assert_eq!(pb_status_to_state(3), SendRequestState::Completed);
255        assert_eq!(pb_status_to_state(4), SendRequestState::Failed);
256        assert_eq!(pb_status_to_state(99), SendRequestState::Unknown);
257    }
258
259    #[test]
260    fn i32_to_u32_saturates_negatives() {
261        assert_eq!(i32_to_u32(-5), 0);
262        assert_eq!(i32_to_u32(0), 0);
263        assert_eq!(i32_to_u32(42), 42);
264    }
265
266    #[test]
267    fn vars_translation_emits_string_values() {
268        let mut vars = std::collections::BTreeMap::new();
269        vars.insert("name".into(), "Daisy".into());
270        let s = vars_to_pb_struct(&vars);
271        assert_eq!(s.fields.len(), 1);
272        let name_v = s.fields.get("name").unwrap();
273        assert_matches!(&name_v.kind, Some(PbValueKind::StringValue(v)) if v == "Daisy");
274    }
275}
pcs_external/client.rs

pcs_external/
client.rs
