zer-cluster 1.1.0

Connected-components clustering and entity storage for the zer entity-resolution library
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

use rusqlite::Connection;
use zer_core::{
    entity::{EntityId, ResolutionMethod},
    error::ZerError,
    record::RecordId,
};

/// Events written to the `resolution_events` table to provide an audit trail
/// for every structural change to the entity store.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub enum ResolutionEvent {
    EntityCreated {
        entity_id: EntityId,
        record_ids: Vec<RecordId>,
    },
    RecordsAdded {
        entity_id: EntityId,
        record_ids: Vec<RecordId>,
        method: ResolutionMethod,
    },
    EntityMerged {
        source_a: EntityId,
        source_b: EntityId,
        into: EntityId,
    },
    EntitySplit {
        source: EntityId,
        into: Vec<EntityId>,
    },
    JudgeApplied {
        entity_id: EntityId,
        pair: (RecordId, RecordId),
        verdict: String,
    },
}

/// Append a provenance event to `resolution_events`.
///
/// Called from `ZalEntityStore` with the locked connection, no additional
/// locking is needed here.
pub fn append_event(conn: &Connection, event: &ResolutionEvent) -> Result<(), ZerError> {
    let (event_type, entity_id, record_ids, score, judge_verdict) = match event {
        ResolutionEvent::EntityCreated {
            entity_id,
            record_ids,
        } => (
            "EntityCreated",
            *entity_id,
            record_ids.clone(),
            None::<f32>,
            None::<String>,
        ),
        ResolutionEvent::RecordsAdded {
            entity_id,
            record_ids,
            ..
        } => ("RecordsAdded", *entity_id, record_ids.clone(), None, None),
        ResolutionEvent::EntityMerged {
            into,
            source_a,
            source_b,
        } => (
            "EntityMerged",
            *into,
            vec![*source_a, *source_b],
            None,
            None,
        ),
        ResolutionEvent::EntitySplit { source, into } => {
            ("EntitySplit", *source, into.clone(), None, None)
        }
        ResolutionEvent::JudgeApplied {
            entity_id,
            pair,
            verdict,
        } => (
            "JudgeApplied",
            *entity_id,
            vec![pair.0, pair.1],
            None,
            Some(verdict.clone()),
        ),
    };

    let ids_json =
        serde_json::to_string(&record_ids).map_err(|e| ZerError::Serialization(e.to_string()))?;
    let now = unix_now();

    conn.execute(
        "INSERT INTO resolution_events
             (event_type, entity_id, record_ids, score, judge_verdict, occurred_at)
         VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
        rusqlite::params![
            event_type,
            entity_id as i64,
            ids_json,
            score,
            judge_verdict,
            now
        ],
    )
    .map_err(|e| ZerError::Store(e.to_string()))?;

    Ok(())
}

pub(crate) fn unix_now() -> i64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs() as i64)
        .unwrap_or(0)
}