ma-core 0.14.0

DIDComm service library: inboxes, outboxes, DID document publishing, and transport abstraction
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
//! DID document publishing to IPFS/IPNS.
//!
//! Provides request/response types, validation, and (with the `kubo` feature)
//! the [`IpfsDidPublisher`] for publishing signed DID documents via the
//! `ma/ipfs/0.0.1` service.

use crate::{Did, Document, Ipld, Message};
use anyhow::{anyhow, Result};
use serde::{Deserialize, Serialize};

pub const MA_IPNS_ALIAS_HASH_PREFIX: &str = "ma-";

#[cfg(all(not(target_arch = "wasm32"), feature = "kubo"))]
use web_time::Duration;

#[cfg(all(not(target_arch = "wasm32"), feature = "kubo"))]
use crate::kubo::{
    dag_put_cbor, import_key, in_flight_pin_name, list_keys, name_publish_with_retry,
    pin_add_named, remote_pin_add_named, wait_for_api, IpnsPublishOptions, PinCleanupRequest,
    PinCleanupScheduler,
};
#[cfg(all(not(target_arch = "wasm32"), feature = "kubo"))]
use reqwest::Url;
#[cfg(all(not(target_arch = "wasm32"), feature = "kubo"))]
use zeroize::Zeroizing;

use crate::service::{MESSAGE_TYPE_IDENTITY_PUBLISH_REQUEST, MESSAGE_TYPE_IPFS_REQUEST};

// ── Wire formats ──────────────────────────────────────────────────────────

/// CBOR payload for `application/vnd.ma.identity.publish.request` messages
/// on `/ma/ipfs/0.0.1`.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct IdentityPublishRequest {
    /// dag-cbor encoded signed [`Document`].
    pub document: Vec<u8>,
    /// Raw 32-byte IPNS signing key (Ed25519 seed). Must be zeroized by receiver.
    pub ipns_secret_key: Vec<u8>,
}

/// CBOR payload for `application/vnd.ma.ipfs.request` messages on
/// `/ma/ipfs/0.0.1`. Receiver replies with the resulting CID.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct IpfsStoreRequest {
    pub content: Vec<u8>,
    pub content_type: String,
}

fn encode_cbor<T: Serialize>(payload: &T) -> Result<Vec<u8>> {
    let mut buf = Vec::new();
    ciborium::ser::into_writer(payload, &mut buf)
        .map_err(|e| anyhow!("failed to encode CBOR payload: {}", e))?;
    Ok(buf)
}

fn sanitize_key_part(part: &str) -> String {
    let mut sanitized = String::new();
    let mut last_was_separator = false;

    for byte in part.bytes() {
        let ch = byte as char;
        if ch.is_ascii_alphanumeric() {
            sanitized.push(ch.to_ascii_lowercase());
            last_was_separator = false;
        } else if ch == '-' || ch == '_' {
            if !last_was_separator && !sanitized.is_empty() {
                sanitized.push(ch);
                last_was_separator = true;
            }
        } else if !last_was_separator && !sanitized.is_empty() {
            sanitized.push('-');
            last_was_separator = true;
        }
    }

    while sanitized.ends_with(['-', '_']) {
        sanitized.pop();
    }

    if sanitized.is_empty() {
        "unknown".to_string()
    } else {
        sanitized
    }
}

fn document_ma_type(document: &Document) -> &str {
    match document.ma.as_ref() {
        Some(Ipld::Map(map)) => match map.get("type") {
            Some(Ipld::String(kind)) => kind,
            _ => "unknown",
        },
        _ => "unknown",
    }
}

/// Build a deterministic Kubo IPNS key name from operator-visible parts.
///
/// The IPNS identity is only exposed as a short blake3 suffix. Callers may add
/// local context such as a runtime slug, while delegated agent publishes can
/// remain anonymous apart from their `ma.type`.
#[must_use]
pub fn ipns_key_name_for_parts(parts: &[&str], ipns_id: &str) -> String {
    let name_parts = if parts.is_empty() {
        vec!["unknown".to_string()]
    } else {
        parts.iter().map(|part| sanitize_key_part(part)).collect()
    };
    let hash = blake3::hash(ipns_id.as_bytes());
    format!(
        "{}{}-{}",
        MA_IPNS_ALIAS_HASH_PREFIX,
        name_parts.join("-"),
        &hash.to_hex()[..16]
    )
}

/// Build the default deterministic Kubo IPNS key name for a DID document.
///
/// Uses `ma.type` when present and falls back to `unknown`.
#[must_use]
pub fn ipns_key_name_for_document(document: &Document) -> String {
    let document_did = Did::try_from(document.id.as_str());
    let ipns_id = document_did
        .as_ref()
        .map_or(document.id.as_str(), |did| did.ipns.as_str());
    ipns_key_name_for_parts(&[document_ma_type(document)], ipns_id)
}

#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct IpfsPublishDidResponse {
    pub ok: bool,
    pub message: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub did: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub cid: Option<String>,
}

#[cfg(all(not(target_arch = "wasm32"), feature = "kubo"))]
#[derive(Clone, Debug)]
/// Publication policy for a native DID document.
pub struct DidDocumentPublishOptions {
    /// Deterministic Kubo key-name components; defaults to the document kind.
    pub key_parts: Vec<String>,
    /// IPNS publication settings.
    pub ipns: IpnsPublishOptions,
    /// Number of bounded retries for IPNS and remote pinning.
    pub attempts: u32,
    /// Initial Fibonacci retry delay.
    pub initial_backoff: Duration,
    /// Optional remote pin service replication policy.
    pub remote_pin: Option<RemotePinOptions>,
    /// Replace older pins with the same name in a background best-effort job.
    pub overwrite: bool,
}

#[cfg(all(not(target_arch = "wasm32"), feature = "kubo"))]
impl Default for DidDocumentPublishOptions {
    fn default() -> Self {
        Self {
            key_parts: Vec::new(),
            ipns: IpnsPublishOptions::default(),
            attempts: 3,
            initial_backoff: Duration::from_secs(1),
            remote_pin: None,
            overwrite: true,
        }
    }
}

#[cfg(all(not(target_arch = "wasm32"), feature = "kubo"))]
#[derive(Clone, Debug, Eq, PartialEq)]
/// Remote Kubo pin service configuration for a published DID document.
pub struct RemotePinOptions {
    /// Kubo pin-service name.
    pub service: String,
    /// Human-readable label for the remote pin.
    pub name: String,
}

#[cfg(all(not(target_arch = "wasm32"), feature = "kubo"))]
#[derive(Clone, Debug, Eq, PartialEq)]
/// Remote replication outcome after local pinning and IPNS publication.
pub enum RemotePinStatus {
    /// The new CID was replicated and stale-pin cleanup was scheduled.
    Replicated { cleanup_scheduled: bool },
    /// Local publication succeeded, but replication failed after retries.
    Degraded { error: String },
}

#[cfg(all(not(target_arch = "wasm32"), feature = "kubo"))]
#[derive(Clone, Debug, Eq, PartialEq)]
/// Confirmed outcome of a locally pinned DID document publication.
pub struct PublishedDidDocument {
    /// CID stored and published through IPNS.
    pub cid: String,
    /// Deterministic Kubo IPNS key alias.
    pub key_name: String,
    /// IPNS identity rooted by the DID.
    pub ipns_id: String,
    /// Kubo accepted the required local recursive pin.
    pub local_pinned: bool,
    /// Whether a detached stale-pin cleanup job was scheduled.
    pub cleanup_scheduled: bool,
    /// Optional remote replication state.
    pub remote_pin: Option<RemotePinStatus>,
}

pub struct ValidatedIdentityPublish {
    pub document_bytes: Vec<u8>,
    pub ipns_secret_key: Vec<u8>,
    pub document: Document,
    pub document_did: Did,
}

/// Validated store request.
pub struct ValidatedIpfsStore {
    pub content: Vec<u8>,
    pub content_type: String,
    pub sender_did: String,
    pub msg_id: String,
}

/// Build CBOR content bytes for `application/vnd.ma.identity.publish.request`.
///
/// The returned bytes are the payload to place in `Message.content` when
/// sending to `/ma/ipfs/0.0.1`.
pub fn generate_identity_publish_request(
    did_document: &Document,
    ipns_secret_key: &[u8],
) -> Result<Vec<u8>> {
    let document_bytes = did_document
        .encode()
        .map_err(|e| anyhow!("failed to encode DID document as dag-cbor: {}", e))?;
    encode_cbor(&IdentityPublishRequest {
        document: document_bytes,
        ipns_secret_key: ipns_secret_key.to_vec(),
    })
}

/// Build a signed `application/vnd.ma.ipfs.request` message (generic store).
///
/// Returns the complete signed [`Message`] ready to send on `/ma/ipfs/0.0.1`.
pub fn generate_ipfs_store_request(
    sender_did: &str,
    publisher_did: &str,
    content: Vec<u8>,
    content_type: &str,
    signing_key: &crate::SigningKey,
) -> Result<Message> {
    let payload = encode_cbor(&IpfsStoreRequest {
        content,
        content_type: content_type.to_string(),
    })?;
    Message::new(
        sender_did,
        publisher_did,
        MESSAGE_TYPE_IPFS_REQUEST,
        "application/cbor",
        &payload,
        signing_key,
    )
    .map_err(|e| anyhow!("failed to build ipfs-store message: {}", e))
}

#[cfg(all(not(target_arch = "wasm32"), feature = "kubo"))]
#[derive(Clone, Debug)]
pub struct IpfsDidPublisher {
    kubo_url: String,
}

#[cfg(all(not(target_arch = "wasm32"), feature = "kubo"))]
impl IpfsDidPublisher {
    pub fn new(kubo_url: impl AsRef<str>) -> Result<Self> {
        let kubo_url = normalize_kubo_url(kubo_url.as_ref())?;
        Ok(Self { kubo_url })
    }

    pub fn kubo_url(&self) -> &str {
        &self.kubo_url
    }

    pub async fn publish_signed_message(
        &self,
        message_cbor: &[u8],
    ) -> Result<IpfsPublishDidResponse> {
        handle_ipfs_publish(&self.kubo_url, message_cbor).await
    }

    pub async fn publish_document(
        &self,
        did_document: Vec<u8>,
        ipns_private_key: Zeroizing<Vec<u8>>,
        options: DidDocumentPublishOptions,
    ) -> Result<PublishedDidDocument> {
        publish_did_document_to_kubo(
            &self.kubo_url,
            PinCleanupScheduler::global(),
            did_document,
            ipns_private_key,
            options,
        )
        .await
    }

    pub async fn wait_until_ready(&self, attempts: u32) -> Result<()> {
        wait_for_api(&self.kubo_url, attempts).await
    }
}

#[cfg(all(not(target_arch = "wasm32"), feature = "kubo"))]
fn normalize_kubo_url(input: &str) -> Result<String> {
    let trimmed = input.trim();
    if trimmed.is_empty() {
        return Err(anyhow!("kubo_url must not be empty"));
    }

    let parsed =
        Url::parse(trimmed).map_err(|e| anyhow!("invalid kubo_url '{}': {}", trimmed, e))?;

    let scheme = parsed.scheme();
    if scheme != "http" && scheme != "https" {
        return Err(anyhow!(
            "kubo_url must use http or https scheme, got '{}'",
            scheme
        ));
    }

    if parsed.host_str().is_none() {
        return Err(anyhow!("kubo_url must include a host"));
    }

    if parsed.query().is_some() || parsed.fragment().is_some() {
        return Err(anyhow!(
            "kubo_url must not include query params or fragments"
        ));
    }

    let mut base = format!("{}://{}", scheme, parsed.host_str().unwrap_or_default());
    if let Some(port) = parsed.port() {
        base.push(':');
        base.push_str(&port.to_string());
    }

    let mut path = parsed.path().trim_end_matches('/').to_string();
    if path.ends_with("/api/v0") {
        path.truncate(path.len() - "/api/v0".len());
    }
    if !path.is_empty() && path != "/" {
        if !path.starts_with('/') {
            base.push('/');
        }
        base.push_str(&path);
    }

    Ok(base)
}

/// Validate a full identity-publish request from raw message CBOR bytes.
///
/// Used internally by [`IpfsDidPublisher::publish_signed_message`].
pub fn validate_identity_publish_request(message_cbor: &[u8]) -> Result<ValidatedIdentityPublish> {
    let message =
        Message::decode(message_cbor).map_err(|e| anyhow!("invalid signed message: {}", e))?;
    validate_identity_publish_message(&message)
}

/// Validate an `application/vnd.ma.identity.publish.request` message.
///
/// Verifies the DID document signature and that the sender IPNS matches the
/// document DID. Returns a [`ValidatedIdentityPublish`].
pub fn validate_identity_publish_message(message: &Message) -> Result<ValidatedIdentityPublish> {
    if message.message_type != MESSAGE_TYPE_IDENTITY_PUBLISH_REQUEST {
        return Err(anyhow!(
            "expected {} on /ma/ipfs/0.0.1, got {}",
            MESSAGE_TYPE_IDENTITY_PUBLISH_REQUEST,
            message.message_type
        ));
    }

    let payload: IdentityPublishRequest =
        ciborium::de::from_reader(message.payload().as_slice())
            .map_err(|e| anyhow!("invalid identity-publish request payload: {}", e))?;
    let IdentityPublishRequest {
        document: document_bytes,
        ipns_secret_key,
    } = payload;

    let sender_did = Did::try_from(message.from.as_str())
        .map_err(|e| anyhow!("invalid sender did '{}': {}", message.from, e))?;

    let document = Document::decode(&document_bytes)
        .map_err(|e| anyhow!("invalid DID document dag-cbor: {}", e))?;
    document
        .validate()
        .map_err(|e| anyhow!("invalid DID document: {}", e))?;
    document
        .verify()
        .map_err(|e| anyhow!("DID document signature verification failed: {}", e))?;

    let document_did = Did::try_from(document.id.as_str())
        .map_err(|e| anyhow!("invalid document DID '{}': {}", document.id, e))?;

    if document_did.ipns != sender_did.ipns {
        return Err(anyhow!(
            "sender IPNS '{}' does not match document IPNS '{}'",
            sender_did.ipns,
            document_did.ipns
        ));
    }

    message
        .verify_with_document(&document)
        .map_err(|e| anyhow!("request signature verification failed: {}", e))?;

    Ok(ValidatedIdentityPublish {
        document_bytes,
        ipns_secret_key,
        document,
        document_did,
    })
}

/// Validate an `application/vnd.ma.ipfs.request` message (generic store).
///
/// Extracts content and sender identity. Returns a [`ValidatedIpfsStore`].
pub fn validate_ipfs_request(message: &Message) -> Result<ValidatedIpfsStore> {
    if message.message_type != MESSAGE_TYPE_IPFS_REQUEST {
        return Err(anyhow!(
            "expected {} on /ma/ipfs/0.0.1, got {}",
            MESSAGE_TYPE_IPFS_REQUEST,
            message.message_type
        ));
    }

    let payload: IpfsStoreRequest = ciborium::de::from_reader(message.payload().as_slice())
        .map_err(|e| anyhow!("invalid IPFS store request payload: {}", e))?;

    Ok(ValidatedIpfsStore {
        content: payload.content,
        content_type: payload.content_type,
        sender_did: message.from.clone(),
        msg_id: message.id.clone(),
    })
}

#[cfg(all(not(target_arch = "wasm32"), feature = "kubo"))]
async fn publish_did_document_to_kubo(
    kubo_url: &str,
    cleanup: &PinCleanupScheduler,
    did_document: Vec<u8>,
    ipns_private_key: Zeroizing<Vec<u8>>,
    options: DidDocumentPublishOptions,
) -> Result<PublishedDidDocument> {
    let document = Document::decode(&did_document)
        .map_err(|e| anyhow!("invalid DID document dag-cbor: {}", e))?;
    let document_did = Did::try_from(document.id.as_str())
        .map_err(|e| anyhow!("invalid document DID '{}': {}", document.id, e))?;
    let document_ipns_id = document_did.ipns.clone();

    let key_parts: Vec<&str> = if options.key_parts.is_empty() {
        vec![document_ma_type(&document)]
    } else {
        options.key_parts.iter().map(String::as_str).collect()
    };
    let key_name = ipns_key_name_for_parts(&key_parts, &document_ipns_id);
    let existing_key = list_keys(kubo_url)
        .await?
        .into_iter()
        .find(|k| k.name == key_name);

    if let Some(existing) = existing_key {
        if existing.id.trim() != document_ipns_id {
            return Err(anyhow!(
                "existing key '{}' has IPNS id '{}' but document DID IPNS is '{}'",
                key_name,
                existing.id,
                document_ipns_id
            ));
        }
    } else {
        if ipns_private_key.is_empty() {
            return Err(anyhow!(
                "ipns_private_key is required when key is not present in Kubo"
            ));
        }

        let raw_key: [u8; 32] = ipns_private_key
            .as_slice()
            .try_into()
            .map_err(|_| anyhow!("ipns_private_key must be 32 bytes"))?;
        let keypair = libp2p_identity::Keypair::ed25519_from_bytes(raw_key)
            .map_err(|e| anyhow!("invalid ipns key: {}", e))?;
        let protobuf_key = keypair
            .to_protobuf_encoding()
            .map_err(|e| anyhow!("failed to encode ipns key: {}", e))?;
        let imported = import_key(kubo_url, &key_name, protobuf_key).await?;
        if imported.id.trim() != document_ipns_id {
            return Err(anyhow!(
                "imported key IPNS id '{}' does not match document DID IPNS '{}'",
                imported.id,
                document_ipns_id
            ));
        }
    }

    let pin_name = options
        .remote_pin
        .as_ref()
        .map(|remote| remote.name.clone())
        .unwrap_or_else(|| key_name.clone());
    // With overwrite, pin under an in-flight name so the fresh pin is safe
    // while the cleanup worker removes stale pins; the worker renames it to
    // the requested name once everything is clean.
    let add_name = if options.overwrite {
        in_flight_pin_name(&pin_name)
    } else {
        pin_name.clone()
    };
    let published_cid = dag_put_cbor(kubo_url, did_document, false).await?;
    pin_add_named(kubo_url, &published_cid, &add_name).await?;
    name_publish_with_retry(
        kubo_url,
        &key_name,
        &document_ipns_id,
        &published_cid,
        &options.ipns,
        options.attempts,
        options.initial_backoff,
    )
    .await?;

    let (cleanup_scheduled, remote_pin) =
        confirm_pins_and_schedule_cleanup(kubo_url, cleanup, pin_name, &published_cid, options)
            .await;

    Ok(PublishedDidDocument {
        cid: published_cid,
        key_name,
        ipns_id: document_ipns_id,
        local_pinned: true,
        cleanup_scheduled,
        remote_pin,
    })
}

#[cfg(all(not(target_arch = "wasm32"), feature = "kubo"))]
async fn confirm_pins_and_schedule_cleanup(
    kubo_url: &str,
    cleanup: &PinCleanupScheduler,
    pin_name: String,
    published_cid: &str,
    options: DidDocumentPublishOptions,
) -> (bool, Option<RemotePinStatus>) {
    let (remote_pin, remote_service) = match options.remote_pin {
        Some(remote) => match remote_pin_with_retry(
            kubo_url,
            &remote,
            published_cid,
            options.overwrite,
            options.attempts,
            options.initial_backoff,
        )
        .await
        {
            Ok(()) => (
                Some(RemotePinStatus::Replicated {
                    cleanup_scheduled: false,
                }),
                Some(remote.service),
            ),
            Err(error) => (
                Some(RemotePinStatus::Degraded {
                    error: error.to_string(),
                }),
                None,
            ),
        },
        None => (None, None),
    };
    let cleanup_scheduled = options.overwrite
        && cleanup.schedule(PinCleanupRequest {
            kubo_url: kubo_url.to_string(),
            name: pin_name,
            protected_cid: published_cid.to_string(),
            cleanup_local: true,
            remote_service,
        });
    let remote_pin = remote_pin.map(|status| match status {
        RemotePinStatus::Replicated { .. } => RemotePinStatus::Replicated { cleanup_scheduled },
        RemotePinStatus::Degraded { error } => RemotePinStatus::Degraded { error },
    });

    (cleanup_scheduled, remote_pin)
}

#[cfg(all(not(target_arch = "wasm32"), feature = "kubo"))]
async fn remote_pin_with_retry(
    kubo_url: &str,
    remote: &RemotePinOptions,
    cid: &str,
    overwrite: bool,
    attempts: u32,
    initial_backoff: Duration,
) -> Result<()> {
    if attempts == 0 {
        return Err(anyhow!("remote pin attempts must be >= 1"));
    }
    // Mirror the local policy: with overwrite the fresh remote pin gets the
    // in-flight name and is renamed by the cleanup worker after the old pins
    // are gone.
    let add_name = if overwrite {
        in_flight_pin_name(&remote.name)
    } else {
        remote.name.clone()
    };
    let mut delay = initial_backoff;
    let mut previous_delay = Duration::ZERO;
    let mut last_error = None;
    for attempt in 1..=attempts {
        match remote_pin_add_named(kubo_url, &remote.service, cid, &add_name).await {
            Ok(()) => return Ok(()),
            Err(error) => last_error = Some(error),
        }
        if attempt < attempts {
            tokio::time::sleep(delay).await;
            let next = previous_delay.saturating_add(delay);
            previous_delay = delay;
            delay = std::cmp::min(next, Duration::from_secs(30));
        }
    }
    Err(last_error.expect("at least one remote pin attempt"))
}

#[cfg(all(not(target_arch = "wasm32"), feature = "kubo"))]
pub async fn handle_ipfs_publish(
    kubo_url: &str,
    message_cbor: &[u8],
) -> Result<IpfsPublishDidResponse> {
    let validated = validate_identity_publish_request(message_cbor)?;

    let published = publish_did_document_to_kubo(
        kubo_url,
        PinCleanupScheduler::global(),
        validated.document_bytes,
        Zeroizing::new(validated.ipns_secret_key),
        DidDocumentPublishOptions::default(),
    )
    .await?;

    Ok(IpfsPublishDidResponse {
        ok: true,
        message: "did document published via ma/ipfs/0.0.1".to_string(),
        did: Some(validated.document_did.id()),
        cid: Some(published.cid),
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{generate_identity_from_secret, Did, MaExtension, SigningKey};

    #[cfg(all(not(target_arch = "wasm32"), feature = "kubo"))]
    use super::normalize_kubo_url;

    fn test_identity(seed: u8) -> crate::GeneratedIdentity {
        generate_identity_from_secret([seed; 32]).expect("identity")
    }

    fn test_signing_key(identity: &crate::GeneratedIdentity) -> SigningKey {
        let sign_url = Did::new_url(&identity.subject_url.ipns, None::<String>).expect("did url");
        let private_key: [u8; 32] = hex::decode(&identity.signing_private_key_hex)
            .expect("decode key")
            .try_into()
            .expect("private key bytes");
        SigningKey::from_private_key_bytes(sign_url, private_key).expect("signing key")
    }

    fn ipfs_url(identity: &crate::GeneratedIdentity) -> String {
        format!("{}#ipfs", identity.document.id)
    }

    fn expected_key_name(parts: &[&str], ipns_id: &str) -> String {
        let hash = blake3::hash(ipns_id.as_bytes());
        format!(
            "{}{}-{}",
            MA_IPNS_ALIAS_HASH_PREFIX,
            parts.join("-"),
            &hash.to_hex()[..16]
        )
    }

    #[test]
    fn document_key_name_uses_ma_type() {
        let identity = test_identity(11);
        let mut document = identity.document.clone();
        document.set_ma_extension(MaExtension::new().kind("agent"));

        assert_eq!(
            ipns_key_name_for_document(&document),
            expected_key_name(&["agent"], &identity.subject_url.ipns)
        );
    }

    #[test]
    fn document_key_name_falls_back_to_unknown_type() {
        let identity = test_identity(12);

        assert_eq!(
            ipns_key_name_for_document(&identity.document),
            expected_key_name(&["unknown"], &identity.subject_url.ipns)
        );
    }

    #[test]
    fn key_name_parts_allow_runtime_slug() {
        let ipns_id = "k51qzi5uqu5example";

        assert_eq!(
            ipns_key_name_for_parts(&["runtime", "my-slug"], ipns_id),
            expected_key_name(&["runtime", "my-slug"], ipns_id)
        );
        assert_eq!(
            ipns_key_name_for_parts(&["runtime", "my-slug", "runtime"], ipns_id),
            expected_key_name(&["runtime", "my-slug", "runtime"], ipns_id)
        );
    }

    #[test]
    fn key_name_parts_are_sanitized() {
        let ipns_id = "k51qzi5uqu5example";

        assert_eq!(
            ipns_key_name_for_parts(&["Runtime", "my slug!", "***"], ipns_id),
            expected_key_name(&["runtime", "my-slug", "unknown"], ipns_id)
        );
    }

    #[test]
    fn generate_request_embeds_cbor_document_and_private_key() {
        let identity = test_identity(21);
        let payload =
            generate_identity_publish_request(&identity.document, b"secret-key").expect("payload");
        let request: IdentityPublishRequest =
            ciborium::de::from_reader(payload.as_slice()).expect("decode request");

        assert_eq!(
            request.document,
            identity.document.encode().expect("document bytes")
        );
        assert_eq!(request.ipns_secret_key, b"secret-key".to_vec());
    }

    #[test]
    fn validate_identity_publish_request_accepts_signed_request() {
        let identity = test_identity(22);
        let signing_key = test_signing_key(&identity);
        let payload =
            generate_identity_publish_request(&identity.document, b"private-key").expect("payload");
        let message = Message::new(
            identity.document.id.clone(),
            ipfs_url(&identity),
            MESSAGE_TYPE_IDENTITY_PUBLISH_REQUEST,
            "application/cbor",
            &payload,
            &signing_key,
        )
        .expect("message");
        let encoded = message.encode().expect("message cbor");

        let validated = validate_identity_publish_request(&encoded).expect("validated request");
        assert_eq!(validated.document, identity.document);
        assert_eq!(validated.ipns_secret_key, b"private-key".to_vec());
    }

    #[test]
    fn validate_identity_publish_request_rejects_wrong_content_type() {
        let identity = test_identity(23);
        let signing_key = test_signing_key(&identity);
        let payload =
            generate_identity_publish_request(&identity.document, b"private-key").expect("payload");
        let message = Message::new(
            identity.document.id.clone(),
            ipfs_url(&identity),
            "application/x-test",
            "application/cbor",
            &payload,
            &signing_key,
        )
        .expect("message");
        let encoded = message.encode().expect("message cbor");

        let err = validate_identity_publish_request(&encoded)
            .err()
            .expect("wrong content type");
        assert!(err
            .to_string()
            .contains("expected application/vnd.ma.identity.publish.request"));
    }

    #[test]
    fn validate_identity_publish_request_rejects_ipns_mismatch() {
        let sender_identity = test_identity(24);
        let document_identity = test_identity(25);
        let signing_key = test_signing_key(&sender_identity);
        let payload =
            generate_identity_publish_request(&document_identity.document, b"private-key")
                .expect("payload");
        let message = Message::new(
            sender_identity.document.id.clone(),
            ipfs_url(&sender_identity),
            MESSAGE_TYPE_IDENTITY_PUBLISH_REQUEST,
            "application/cbor",
            &payload,
            &signing_key,
        )
        .expect("message");
        let encoded = message.encode().expect("message cbor");

        let err = validate_identity_publish_request(&encoded)
            .err()
            .expect("ipns mismatch");
        assert!(err.to_string().contains("does not match document IPNS"));
    }

    #[test]
    fn validate_identity_publish_request_rejects_invalid_document_bytes() {
        let identity = test_identity(26);
        let signing_key = test_signing_key(&identity);
        let payload = encode_cbor(&IdentityPublishRequest {
            document: b"not dag-cbor".to_vec(),
            ipns_secret_key: b"private-key".to_vec(),
        })
        .expect("encode request");
        let message = Message::new(
            identity.document.id.clone(),
            ipfs_url(&identity),
            MESSAGE_TYPE_IDENTITY_PUBLISH_REQUEST,
            "application/cbor",
            &payload,
            &signing_key,
        )
        .expect("message");
        let encoded = message.encode().expect("message cbor");

        let err = validate_identity_publish_request(&encoded)
            .err()
            .expect("invalid document");
        assert!(
            err.to_string()
                .contains("invalid identity-publish request payload")
                || err.to_string().contains("invalid DID document dag-cbor")
        );
    }

    #[test]
    fn validate_identity_publish_request_rejects_malformed_document_shapes() {
        let identity = test_identity(27);
        let signing_key = test_signing_key(&identity);

        let mut wrong_context = identity.document.clone();
        wrong_context.context = vec!["https://www.w3.org/ns/did/v1".to_string()];

        let mut fragmented_id = identity.document.clone();
        fragmented_id.id.push_str("#subject");

        let mut fragmented_controller = identity.document.clone();
        fragmented_controller.controller[0].push_str("#controller");

        let mut wrong_method_type = identity.document.clone();
        wrong_method_type.verification_method[0].key_type = "JsonWebKey2020".to_string();

        let mut missing_relationship_target = identity.document.clone();
        missing_relationship_target.assertion_method[0] =
            format!("{}#unknown", missing_relationship_target.id);

        let mut wrong_assertion_codec = identity.document.clone();
        wrong_assertion_codec.assertion_method[0] = wrong_assertion_codec.key_agreement[0].clone();

        let mut wrong_agreement_codec = identity.document.clone();
        wrong_agreement_codec.key_agreement[0] = wrong_agreement_codec.assertion_method[0].clone();

        for (name, document) in [
            ("wrong context", wrong_context),
            ("fragmented document id", fragmented_id),
            ("fragmented controller", fragmented_controller),
            ("wrong verification method type", wrong_method_type),
            ("missing relationship target", missing_relationship_target),
            ("wrong assertion codec", wrong_assertion_codec),
            ("wrong key-agreement codec", wrong_agreement_codec),
        ] {
            let payload = generate_identity_publish_request(&document, b"private-key")
                .expect("publish payload");
            let message = Message::new(
                identity.document.id.clone(),
                ipfs_url(&identity),
                MESSAGE_TYPE_IDENTITY_PUBLISH_REQUEST,
                "application/cbor",
                &payload,
                &signing_key,
            )
            .expect("message");

            let err = validate_identity_publish_request(&message.encode().expect("message cbor"))
                .err()
                .unwrap_or_else(|| panic!("accepted malformed document: {name}"));
            assert!(
                err.to_string().contains("invalid DID document"),
                "unexpected error for {name}: {err}"
            );
        }
    }

    #[test]
    fn validate_identity_publish_request_rejects_invalid_proof_metadata() {
        let identity = test_identity(28);
        let signing_key = test_signing_key(&identity);

        let mut wrong_type = identity.document.clone();
        wrong_type.proof.proof_type = "DataIntegrityProof".to_string();

        let mut wrong_purpose = identity.document.clone();
        wrong_purpose.proof.proof_purpose = "authentication".to_string();

        for (name, document) in [("proof type", wrong_type), ("proof purpose", wrong_purpose)] {
            let payload = generate_identity_publish_request(&document, b"private-key")
                .expect("publish payload");
            let message = Message::new(
                identity.document.id.clone(),
                ipfs_url(&identity),
                MESSAGE_TYPE_IDENTITY_PUBLISH_REQUEST,
                "application/cbor",
                &payload,
                &signing_key,
            )
            .expect("message");

            let err = validate_identity_publish_request(&message.encode().expect("message cbor"))
                .err()
                .unwrap_or_else(|| panic!("accepted invalid {name}"));
            assert!(
                err.to_string()
                    .contains("DID document signature verification failed"),
                "unexpected error for {name}: {err}"
            );
        }
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "kubo"))]
    #[test]
    fn normalizes_trailing_slash() {
        assert_eq!(
            normalize_kubo_url("http://127.0.0.1:5001/").expect("normalize url"),
            "http://127.0.0.1:5001"
        );
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "kubo"))]
    #[test]
    fn strips_api_v0_suffix() {
        assert_eq!(
            normalize_kubo_url("http://127.0.0.1:5001/api/v0").expect("normalize url"),
            "http://127.0.0.1:5001"
        );
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "kubo"))]
    #[test]
    fn keeps_custom_base_path() {
        assert_eq!(
            normalize_kubo_url("http://localhost:5001/kubo").expect("normalize url"),
            "http://localhost:5001/kubo"
        );
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "kubo"))]
    #[test]
    fn rejects_empty_url() {
        assert!(normalize_kubo_url("   ").is_err());
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "kubo"))]
    #[test]
    fn rejects_non_http_scheme() {
        assert!(normalize_kubo_url("ftp://127.0.0.1:5001").is_err());
    }
}