dwn 0.4.0

Rust implementation of a Decentralized Web Node.
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

use dwn_core::{message::descriptor::Descriptor, reply::RecordsSyncReply, store::StoreError};
use reqwest::StatusCode;
use tracing::warn;

use crate::ProcessContext;

pub async fn handle(
    ProcessContext {
        rs,
        ds,
        validation,
        target,
        msg,
    }: ProcessContext<'_>,
) -> Result<RecordsSyncReply, StatusCode> {
    debug_assert!(matches!(msg.descriptor, Descriptor::RecordsSync(_)));

    let Descriptor::RecordsSync(desc) = msg.descriptor else {
        panic!("invalid descriptor: {:?}", msg.descriptor);
    };

    let authorized = validation.authenticated.contains(target);

    let mut reply = RecordsSyncReply {
        conflict: Vec::new(),
        local_only: Vec::new(),
        remote_only: Vec::new(),
    };

    let mut local = rs.prepare_sync(target, authorized).map_err(|e| {
        warn!("Failed to prepare sync {}: {:?}", msg.record_id, e);
        StatusCode::INTERNAL_SERVER_ERROR
    })?;

    for record in desc.local_records {
        // Remove from local records.
        if let Some(found_idx) = local
            .local_records
            .iter()
            .position(|v| v.record_id == record.record_id)
        {
            local.local_records.remove(found_idx);
        };

        // Process given record.
        if let Some(found) = rs.read(ds, target, &record.record_id).map_err(|e| {
            warn!("Failed to read record {}: {:?}", msg.record_id, e);
            StatusCode::INTERNAL_SERVER_ERROR
        })? {
            if found
                .latest_entry
                .descriptor
                .compute_entry_id()
                .map_err(|e| {
                    warn!("Failed to compute entry id {}: {:?}", msg.record_id, e);
                    StatusCode::INTERNAL_SERVER_ERROR
                })?
                != record.latest_entry_id
            {
                reply.conflict.push(found.latest_entry);
            }
        } else {
            reply.local_only.push(record.record_id);
        };
    }

    reply.remote_only = local
        .local_records
        .into_iter()
        .map(|id| match rs.read(ds, target, &id.record_id) {
            Ok(Some(r)) => Ok(r),
            Ok(None) => Err(StoreError::BackendError(
                "Sync record not found".to_string(),
            )),
            Err(e) => Err(e),
        })
        .collect::<Result<Vec<_>, _>>()
        .map_err(|e| {
            warn!(
                "Failed to read record {} during sync: {:?}",
                msg.record_id, e
            );
            StatusCode::INTERNAL_SERVER_ERROR
        })?;

    Ok(reply)
}