paykit_lib/encrypted_link/
private_application_message.rs

1use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine as _};
2use pubky::{errors::RequestError, Error as PubkyError, PubkySession, PublicKey, StatusCode};
3use tracing::{debug, warn};
4
5use crate::{
6    encrypted_link::paths::compute_private_payment_paths,
7    error::{NonRetryablePrivateReceiveError, NonRetryablePrivateSendError},
8    PaykitError, Result,
9};
10
11// Local marker used when decrypted private plaintext is not valid UTF-8.
12// This is not a protocol message. It lets receive callers persist the malformed
13// stream item, advance their local link checkpoint, and avoid wedging on the
14// same encrypted slot forever.
15const INVALID_UTF8_PRIVATE_MESSAGE_PREFIX: &str = "paykit.invalid_utf8_private_message:";
16const LIST_PAGE_LIMIT: u16 = 100;
17/// Maximum encrypted stream slots consumed by one receive call.
18pub const PRIVATE_APPLICATION_MESSAGE_RECEIVE_LIMIT: usize = 100;
19
20/// Private Message Kind values understood by Paykit.
21///
22/// This enum is intentionally exhaustive. Adding a variant must produce
23/// compile-time failures in routing, validation, and backup-validation matches
24/// until the new Private Message Kind is classified explicitly. Do not add
25/// `#[non_exhaustive]` without a coordinated team decision.
26#[derive(Clone, Copy, Debug, PartialEq, Eq)]
27pub enum PrivateMessageKind {
28    /// Private Payment List Latest-State Message (`paykit.private_payment_list`).
29    PrivatePaymentList,
30    /// Delivery Confirmation (`paykit.delivery_confirmation`), not an Event Message.
31    DeliveryConfirmation,
32    /// Receipt Access Event Message (`paykit.receipt_access`).
33    ReceiptAccess,
34    /// Payment Request Event Message (`paykit.payment_request`).
35    PaymentRequest,
36    /// Payment Request Acceptance Event Message (`paykit.payment_request_acceptance`).
37    PaymentRequestAcceptance,
38    /// Payment Request Rejection Event Message (`paykit.payment_request_rejection`).
39    PaymentRequestRejection,
40    /// Payment Request Cancellation Event Message (`paykit.payment_request_cancellation`).
41    PaymentRequestCancellation,
42    /// Recurring Payment Conversion Quote Event Message (`paykit.payment_conversion_quote`).
43    PaymentConversionQuote,
44    /// Payment Proof Event Message (`paykit.payment_proof`).
45    PaymentProof,
46    /// Allowance Proposal Event Message (`paykit.allowance_proposal`).
47    AllowanceProposal,
48    /// Allowance Acceptance Event Message (`paykit.allowance_acceptance`).
49    AllowanceAcceptance,
50    /// Allowance Rejection Event Message (`paykit.allowance_rejection`).
51    AllowanceRejection,
52    /// Allowance End Event Message (`paykit.allowance_end`).
53    AllowanceEnd,
54}
55
56impl PrivateMessageKind {
57    /// Return the canonical private message kind string used on the wire.
58    pub fn as_str(self) -> &'static str {
59        match self {
60            Self::PrivatePaymentList => "paykit.private_payment_list",
61            Self::DeliveryConfirmation => "paykit.delivery_confirmation",
62            Self::ReceiptAccess => "paykit.receipt_access",
63            Self::PaymentRequest => "paykit.payment_request",
64            Self::PaymentRequestAcceptance => "paykit.payment_request_acceptance",
65            Self::PaymentRequestRejection => "paykit.payment_request_rejection",
66            Self::PaymentRequestCancellation => "paykit.payment_request_cancellation",
67            Self::PaymentProof => "paykit.payment_proof",
68            Self::PaymentConversionQuote => "paykit.payment_conversion_quote",
69            Self::AllowanceProposal => "paykit.allowance_proposal",
70            Self::AllowanceAcceptance => "paykit.allowance_acceptance",
71            Self::AllowanceRejection => "paykit.allowance_rejection",
72            Self::AllowanceEnd => "paykit.allowance_end",
73        }
74    }
75
76    /// Parse a canonical private message kind string.
77    pub fn parse(kind: &str) -> Option<Self> {
78        match kind {
79            "paykit.private_payment_list" => Some(Self::PrivatePaymentList),
80            "paykit.delivery_confirmation" => Some(Self::DeliveryConfirmation),
81            "paykit.receipt_access" => Some(Self::ReceiptAccess),
82            "paykit.payment_request" => Some(Self::PaymentRequest),
83            "paykit.payment_request_acceptance" => Some(Self::PaymentRequestAcceptance),
84            "paykit.payment_request_rejection" => Some(Self::PaymentRequestRejection),
85            "paykit.payment_request_cancellation" => Some(Self::PaymentRequestCancellation),
86            "paykit.payment_proof" => Some(Self::PaymentProof),
87            "paykit.payment_conversion_quote" => Some(Self::PaymentConversionQuote),
88            "paykit.allowance_proposal" => Some(Self::AllowanceProposal),
89            "paykit.allowance_acceptance" => Some(Self::AllowanceAcceptance),
90            "paykit.allowance_rejection" => Some(Self::AllowanceRejection),
91            "paykit.allowance_end" => Some(Self::AllowanceEnd),
92            _ => None,
93        }
94    }
95
96    /// Whether this kind carries an Event Message with its own Event ID.
97    ///
98    /// Delivery Confirmations refer to an existing event and must not themselves
99    /// receive confirmations. Private Payment Lists use Latest-State semantics.
100    pub fn is_event(self) -> bool {
101        match self {
102            Self::ReceiptAccess
103            | Self::PaymentRequest
104            | Self::PaymentRequestAcceptance
105            | Self::PaymentRequestRejection
106            | Self::PaymentRequestCancellation
107            | Self::PaymentConversionQuote
108            | Self::AllowanceProposal
109            | Self::AllowanceAcceptance
110            | Self::AllowanceRejection
111            | Self::AllowanceEnd
112            | Self::PaymentProof => true,
113            Self::PrivatePaymentList | Self::DeliveryConfirmation => false,
114        }
115    }
116}
117
118impl std::fmt::Display for PrivateMessageKind {
119    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
120        f.write_str(self.as_str())
121    }
122}
123
124/// One Private Application Message received from an Encrypted Link.
125///
126/// This is the low-level receive item for the Private Application Message stream.
127/// It keeps the raw plaintext payload so callers can route and persist messages
128/// themselves. If decrypted bytes are not UTF-8, `raw_json` contains a local
129/// invalid-JSON marker with the bytes encoded as base64url. Higher layers should
130/// treat that marker as malformed input, not as a Private Application Message.
131#[derive(Clone, PartialEq, Eq)]
132pub struct PrivateApplicationMessage {
133    /// Private Application Message version from the JSON `version` field, when
134    /// the field is present and can be represented as a `u8`.
135    pub version: Option<u8>,
136    /// Message kind string from the JSON `kind` field, when the field is a
137    /// string.
138    pub kind: Option<String>,
139    /// App ID string from the JSON `app_id` field, when the field is a string.
140    pub app_id: Option<String>,
141    /// Raw plaintext received over the Encrypted Link.
142    pub raw_json: String,
143}
144
145impl std::fmt::Debug for PrivateApplicationMessage {
146    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
147        f.debug_struct("PrivateApplicationMessage")
148            .field("version", &self.version)
149            .field("kind", &self.kind)
150            .field("app_id", &self.app_id)
151            .field(
152                "raw_json",
153                &format_args!("<redacted:{} bytes>", self.raw_json.len()),
154            )
155            .finish()
156    }
157}
158
159impl PrivateApplicationMessage {
160    pub(super) fn from_plaintext(plaintext: String) -> Self {
161        let value = serde_json::from_str::<serde_json::Value>(&plaintext).ok();
162        let version = value
163            .as_ref()
164            .and_then(|value| value.get("version"))
165            .and_then(serde_json::Value::as_u64)
166            .and_then(|version| u8::try_from(version).ok());
167        let kind = value
168            .as_ref()
169            .and_then(|value| value.get("kind"))
170            .and_then(serde_json::Value::as_str)
171            .map(str::to_owned);
172        let app_id = value
173            .as_ref()
174            .and_then(|value| value.get("app_id"))
175            .and_then(serde_json::Value::as_str)
176            .map(str::to_owned);
177        Self {
178            version,
179            kind,
180            app_id,
181            raw_json: plaintext,
182        }
183    }
184
185    /// Return the known Paykit kind from the raw payload, if this library can
186    /// parse and recognize it.
187    pub fn known_kind(&self) -> Option<PrivateMessageKind> {
188        serde_json::from_str::<serde_json::Value>(&self.raw_json)
189            .ok()
190            .and_then(|value| {
191                value
192                    .get("kind")
193                    .and_then(serde_json::Value::as_str)
194                    .and_then(PrivateMessageKind::parse)
195            })
196    }
197
198    /// Return a parse error for the local invalid-UTF-8 receive marker.
199    ///
200    /// SDK/runtime callers use this to persist an audit/error record for a
201    /// malformed stream item while still advancing the local Encrypted Link
202    /// checkpoint. This is not expected on valid protocol messages.
203    pub fn invalid_utf8_error(&self) -> Option<&'static str> {
204        self.raw_json
205            .starts_with(INVALID_UTF8_PRIVATE_MESSAGE_PREFIX)
206            .then_some("Private Application Message plaintext is not valid UTF-8")
207    }
208}
209
210pub(super) fn decode_private_application_message(raw: &[u8]) -> Result<PrivateApplicationMessage> {
211    // pubky-noise returns the exact authenticated application body. It owns
212    // transport framing and padding; preserve malformed JSON for caller routing.
213    let plaintext = match std::str::from_utf8(raw) {
214        Ok(plaintext) => plaintext.to_owned(),
215        Err(_) => format!(
216            "{INVALID_UTF8_PRIVATE_MESSAGE_PREFIX}{}",
217            URL_SAFE_NO_PAD.encode(raw)
218        ),
219    };
220
221    Ok(PrivateApplicationMessage::from_plaintext(plaintext))
222}
223
224pub(super) async fn receive_private_application_messages(
225    encryptor: &mut pubky_noise::PubkyNoiseEncryptor,
226) -> Result<Vec<PrivateApplicationMessage>> {
227    let mut messages = Vec::new();
228
229    while messages.len() < PRIVATE_APPLICATION_MESSAGE_RECEIVE_LIMIT {
230        let batch = encryptor.receive_message().await.map_err(|err| {
231            let context = format!("failed to receive Private Application Messages: {err:?}");
232            let source = if is_retryable_private_receive_error(&err) {
233                anyhow::anyhow!("pubky-noise receive_message failed: {err:?}")
234            } else {
235                anyhow::Error::new(NonRetryablePrivateReceiveError(err))
236            };
237            PaykitError::Transport { context, source }
238        })?;
239
240        if batch.is_empty() {
241            break;
242        }
243
244        for raw in batch {
245            messages.push(decode_private_application_message(&raw)?);
246        }
247    }
248
249    Ok(messages)
250}
251
252/// Delete all encrypted stream slots written by the local identity for one
253/// counterparty and recovery context.
254///
255/// This clears the local write path used by the counterparty as their read path.
256/// Only retire a context after neither peer needs its handshake or messages.
257/// Recovery itself uses a new context and does not require deletion. The
258/// counterparty's outbox is not touched. The local Pubky identity comes from `session`; the remote
259/// identity and Noise key must belong to the intended counterparty. The caller
260/// owns session creation, capability scope, key rotation, and request timeouts.
261pub async fn clear_encrypted_link_outbox(
262    session: &PubkySession,
263    local_secret_key: &[u8; 32],
264    remote_identity_public_key: &PublicKey,
265    remote_noise_public_key: &PublicKey,
266    recovery_context: &super::EncryptedLinkRecoveryContext,
267) -> Result<usize> {
268    let (write_path, _) = compute_private_payment_paths(
269        local_secret_key,
270        session.info().public_key(),
271        remote_identity_public_key,
272        remote_noise_public_key,
273        recovery_context,
274    );
275    let list_path = format!("{write_path}/");
276    let storage = session.storage();
277    let mut deleted_count = 0;
278
279    loop {
280        let builder = match storage.list(&list_path) {
281            Ok(builder) => builder.shallow(true).limit(LIST_PAGE_LIMIT),
282            Err(err) if is_not_found(&err) => return Ok(deleted_count),
283            Err(err) => {
284                return Err(PaykitError::Transport {
285                    context: "list Encrypted Link outbox".into(),
286                    source: err.into(),
287                });
288            }
289        };
290
291        let page = match builder.send().await {
292            Ok(page) => page,
293            Err(err) if is_not_found(&err) => return Ok(deleted_count),
294            Err(err) => {
295                return Err(PaykitError::Transport {
296                    context: "list Encrypted Link outbox".into(),
297                    source: err.into(),
298                });
299            }
300        };
301
302        if page.is_empty() {
303            return Ok(deleted_count);
304        }
305
306        let page_len = page.len();
307        for resource in page {
308            match storage.delete(resource.path.as_str()).await {
309                Ok(_) => deleted_count += 1,
310                Err(err) if is_not_found(&err) => {}
311                Err(err) => {
312                    return Err(PaykitError::Transport {
313                        context: "clear Encrypted Link outbox".into(),
314                        source: err.into(),
315                    });
316                }
317            }
318        }
319
320        if page_len < LIST_PAGE_LIMIT as usize {
321            return Ok(deleted_count);
322        }
323    }
324}
325
326fn send_attempts_from_retries(max_send_retries: u32) -> u32 {
327    max_send_retries.saturating_add(1)
328}
329
330fn is_retryable_private_send_error(err: &pubky_noise::PubkyNoiseError) -> bool {
331    matches!(err, pubky_noise::PubkyNoiseError::HomeserverWriteError)
332}
333
334fn is_retryable_private_receive_error(err: &pubky_noise::PubkyNoiseError) -> bool {
335    // HTTP failures do not establish corrupt link state. Retry policy stays
336    // with the caller, including authentication and rate-limit failures.
337    matches!(
338        err,
339        pubky_noise::PubkyNoiseError::HomeserverResponseError
340            | pubky_noise::PubkyNoiseError::HomeserverHttpError { .. }
341    )
342}
343
344fn is_not_found(err: &PubkyError) -> bool {
345    matches!(
346        err,
347        PubkyError::Request(RequestError::Server { status, .. })
348            if *status == StatusCode::NOT_FOUND || *status == StatusCode::GONE
349    )
350}
351
352pub(super) fn validate_private_application_message_size(
353    plaintext: &[u8],
354    context: &'static str,
355) -> Result<()> {
356    if plaintext.len() > pubky_noise::snow_crypto::PUBKY_NOISE_MSG_LEN {
357        return Err(PaykitError::Validation(format!(
358            "{context} payload ({} bytes) exceeds max message size ({} bytes)",
359            plaintext.len(),
360            pubky_noise::snow_crypto::PUBKY_NOISE_MSG_LEN,
361        )));
362    }
363    Ok(())
364}
365
366#[allow(
367    deprecated,
368    reason = "the stateless convenience API cannot persist a prepared send"
369)]
370pub(super) async fn send_private_application_message(
371    encryptor: &mut pubky_noise::PubkyNoiseEncryptor,
372    max_send_retries: u32,
373    plaintext: &[u8],
374    context: &'static str,
375) -> Result<()> {
376    validate_private_application_message_size(plaintext, context)?;
377
378    let max_attempts = send_attempts_from_retries(max_send_retries);
379    let mut last_error: Option<String> = None;
380    // A failed outbox write leaves the prepared packet unacknowledged.
381    // Preparing a new send would then fail with
382    // `UnacknowledgedPreparedTransport` (non-retryable), so retries republish
383    // the same prepared packet instead of preparing a new one.
384    let mut retry_pending = false;
385
386    for attempt in 1..=max_attempts {
387        let outcome = if retry_pending {
388            encryptor.retry_pending_send().await
389        } else {
390            encryptor.send_message(plaintext).await
391        };
392        match outcome {
393            Ok(()) => {
394                debug!(context, "Private Application Message sent successfully");
395                return Ok(());
396            }
397            Err(err) if is_retryable_private_send_error(&err) => {
398                retry_pending = true;
399                last_error = Some(format!("{err:?}"));
400                if attempt < max_attempts {
401                    warn!(
402                        attempt,
403                        max_retries = max_send_retries,
404                        error = ?err,
405                        context,
406                        "send_message failed, retrying"
407                    );
408                }
409            }
410            Err(err) => {
411                let context = format!("failed to send {context}: {err:?}");
412                return Err(PaykitError::Transport {
413                    context,
414                    source: anyhow::Error::new(NonRetryablePrivateSendError(err)),
415                });
416            }
417        }
418    }
419
420    Err(PaykitError::Transport {
421        context: format!("failed to send {context} after {max_attempts} attempts"),
422        source: anyhow::anyhow!(
423            "pubky-noise send_message failed on all {} attempts; last error: {}",
424            max_attempts,
425            last_error.unwrap_or_else(|| "unknown error".to_string())
426        ),
427    })
428}
429
430#[cfg(test)]
431mod tests;
paykit_lib/encrypted_link/private_application_message.rs

paykit_lib/encrypted_link/
private_application_message.rs
