OXYGENHUB
# f4e138301198 oxygenhub / crates / hub / src / db / tlog.rs
4.3 KiB · Rust · last changed Trusted time on the hub: serve when witnesses first saw an entry 17 days ago source blame history raw
1 
//! The transparency log's index (migration 0076): which entry logged which
2 
//! object. The log itself is files (`crate::tlog`).
3
4 
use super::Pool;
5 
use sqlx::FromRow;
6
7 
#[derive(Clone, Debug, FromRow)]
8 
pub struct Logged {
9 
    pub idx: i64,
10 
    pub object: String,
11 
    pub kind: String,
12 
    pub repo_id: Option<i64>,
13 
    pub received_ms: i64,
14 
}
15
16 
/// The entry that logged `object` as `kind`, if one did.
17 
pub async fn entry_of(pool: &Pool, object: &str, kind: &str) -> sqlx::Result<Option<i64>> {
18 
    sqlx::query_scalar("SELECT idx FROM tlog_entries WHERE object = $1 AND kind = $2").bind(object).bind(kind).fetch_optional(pool).await
19 
}
20
21 
pub async fn record(pool: &Pool, idx: i64, object: &str, kind: &str, repo_id: Option<i64>, received_ms: i64) -> sqlx::Result<()> {
22 
    sqlx::query("INSERT INTO tlog_entries (idx, object, kind, repo_id, received_ms) VALUES ($1, $2, $3, $4, $5) ON CONFLICT DO NOTHING")
23 
        .bind(idx)
24 
        .bind(object)
25 
        .bind(kind)
26 
        .bind(repo_id)
27 
        .bind(received_ms)
28 
        .execute(pool)
29 
        .await?;
30 
    Ok(())
31 
}
32
33 
/// The newest entries, newest first; of one repository when given.
34 
pub async fn recent(pool: &Pool, repo_id: Option<i64>, limit: i64) -> sqlx::Result<Vec<Logged>> {
35 
    match repo_id {
36 
        Some(r) => {
37 
            sqlx::query_as("SELECT * FROM tlog_entries WHERE repo_id = $1 ORDER BY idx DESC LIMIT $2")
38 
                .bind(r)
39 
                .bind(limit)
40 
                .fetch_all(pool)
41 
                .await
42 
        }
43 
        None => sqlx::query_as("SELECT * FROM tlog_entries ORDER BY idx DESC LIMIT $1").bind(limit).fetch_all(pool).await,
44 
    }
45 
}
46
47 
/// The last checkpoint this hub cosigned for log `origin`: its size and
48 
/// root (hex).
49 
pub async fn witnessed(pool: &Pool, origin: &str) -> sqlx::Result<Option<(i64, String)>> {
50 
    sqlx::query_as("SELECT size, root FROM tlog_witnessed WHERE origin = $1").bind(origin).fetch_optional(pool).await
51 
}
52
53 
/// Move what this hub cosigned for `origin` from `old` entries (none when
54 
/// zero) to `size` with `root`; false when another request moved it first.
55 
pub async fn witness_advance(pool: &Pool, origin: &str, old: i64, size: i64, root: &str) -> sqlx::Result<bool> {
56 
    let now = super::now_ms();
57 
    let r = if old == 0 {
58 
        sqlx::query(
59 
            "INSERT INTO tlog_witnessed (origin, size, root, updated_ms) VALUES ($1, $2, $3, $4) \
60 
             ON CONFLICT (origin) DO UPDATE SET size = $2, root = $3, updated_ms = $4 WHERE tlog_witnessed.size = 0",
61 
        )
62 
        .bind(origin)
63 
        .bind(size)
64 
        .bind(root)
65 
        .bind(now)
66 
        .execute(pool)
67 
        .await?
68 
    } else {
69 
        sqlx::query("UPDATE tlog_witnessed SET size = $1, root = $2, updated_ms = $3 WHERE origin = $4 AND size = $5")
70 
            .bind(size)
71 
            .bind(root)
72 
            .bind(now)
73 
            .bind(origin)
74 
            .bind(old)
75 
            .execute(pool)
76 
            .await?
77 
    };
78 
    Ok(r.rows_affected() == 1)
79 
}
80
81 
/// How far witness `name` has cosigned this hub's log.
82 
pub async fn cosigned_size(pool: &Pool, name: &str) -> sqlx::Result<i64> {
83 
    Ok(sqlx::query_scalar("SELECT size FROM tlog_cosigned WHERE witness = $1").bind(name).fetch_optional(pool).await?.unwrap_or(0))
84 
}
85
86 
pub async fn set_cosigned(pool: &Pool, name: &str, size: i64) -> sqlx::Result<()> {
87 
    sqlx::query(
88 
        "INSERT INTO tlog_cosigned (witness, size, updated_ms) VALUES ($1, $2, $3) \
89 
         ON CONFLICT (witness) DO UPDATE SET size = $2, updated_ms = $3",
90 
    )
91 
    .bind(name)
92 
    .bind(size)
93 
    .bind(super::now_ms())
94 
    .execute(pool)
95 
    .await?;
96 
    Ok(())
97 
}
98
99 
/// Keep the cosigned checkpoint of `size` entries, with every cosignature
100 
/// it has gathered so far (#240).
101 
pub async fn keep_cosigned_note(pool: &Pool, size: i64, note: &str) -> sqlx::Result<()> {
102 
    sqlx::query(
103 
        "INSERT INTO tlog_cosigned_notes (size, note, updated_ms) VALUES ($1, $2, $3) \
104 
         ON CONFLICT (size) DO UPDATE SET note = $2, updated_ms = $3",
105 
    )
106 
    .bind(size)
107 
    .bind(note)
108 
    .bind(super::now_ms())
109 
    .execute(pool)
110 
    .await?;
111 
    Ok(())
112 
}
113
114 
/// The first `n` cosigned checkpoints that hold entry `index`: (size, note).
115 
pub async fn cosigned_notes_after(pool: &Pool, index: i64, n: i64) -> sqlx::Result<Vec<(i64, String)>> {
116 
    sqlx::query_as("SELECT size, note FROM tlog_cosigned_notes WHERE size > $1 ORDER BY size LIMIT $2")
117 
        .bind(index)
118 
        .bind(n)
119 
        .fetch_all(pool)
120 
        .await
121 
}