OXYGENHUB
# 2c47eecfeab6 oxygenhub / crates / hub / src / db / terraform.rs
3.9 KiB · Rust · last changed The hub keeps Terraform and OpenTofu state (closes #252) 18 days ago source blame history raw
1 
//! Terraform and OpenTofu states (migration 0079, oxygenhub#252).
2
3 
use super::{now_ms, Pool};
4 
use sqlx::FromRow;
5
6 
#[derive(Clone, Debug, FromRow)]
7 
pub struct TfState {
8 
    pub repo_id: i64,
9 
    pub name: String,
10 
    /// Sealed with the hub's secrets key; empty once deleted.
11 
    pub body: Vec<u8>,
12 
    pub serial: i64,
13 
    pub lock_id: String,
14 
    pub lock_info: String,
15 
    pub locked_by: String,
16 
    pub locked_ms: Option<i64>,
17 
    pub updated_by: String,
18 
    pub updated_ms: i64,
19 
}
20
21 
pub async fn get(pool: &Pool, repo_id: i64, name: &str) -> sqlx::Result<Option<TfState>> {
22 
    sqlx::query_as("SELECT * FROM terraform_states WHERE repo_id = $1 AND name = $2").bind(repo_id).bind(name).fetch_optional(pool).await
23 
}
24
25 
pub async fn of_repo(pool: &Pool, repo_id: i64) -> sqlx::Result<Vec<TfState>> {
26 
    sqlx::query_as("SELECT * FROM terraform_states WHERE repo_id = $1 ORDER BY name").bind(repo_id).fetch_all(pool).await
27 
}
28
29 
/// How many earlier versions a state keeps.
30 
pub async fn versions(pool: &Pool, repo_id: i64, name: &str) -> sqlx::Result<i64> {
31 
    sqlx::query_scalar("SELECT COUNT(*) FROM terraform_state_versions WHERE repo_id = $1 AND name = $2")
32 
        .bind(repo_id)
33 
        .bind(name)
34 
        .fetch_one(pool)
35 
        .await
36 
}
37
38 
/// Store a new body; the one it replaces becomes a version, and only the
39 
/// last `keep` versions stay.
40 
pub async fn put(pool: &Pool, repo_id: i64, name: &str, body: &[u8], serial: i64, by: &str, keep: i64) -> sqlx::Result<()> {
41 
    let now = now_ms();
42 
    let mut tx = pool.begin().await?;
43 
    sqlx::query(
44 
        "INSERT INTO terraform_state_versions (repo_id, name, body, serial, created_by, created_ms) \
45 
         SELECT repo_id, name, body, serial, updated_by, updated_ms FROM terraform_states \
46 
         WHERE repo_id = $1 AND name = $2 AND length(body) > 0",
47 
    )
48 
    .bind(repo_id)
49 
    .bind(name)
50 
    .execute(&mut *tx)
51 
    .await?;
52 
    sqlx::query(
53 
        "INSERT INTO terraform_states (repo_id, name, body, serial, updated_by, updated_ms) VALUES ($1, $2, $3, $4, $5, $6) \
54 
         ON CONFLICT (repo_id, name) DO UPDATE SET body = $3, serial = $4, updated_by = $5, updated_ms = $6",
55 
    )
56 
    .bind(repo_id)
57 
    .bind(name)
58 
    .bind(body)
59 
    .bind(serial)
60 
    .bind(by)
61 
    .bind(now)
62 
    .execute(&mut *tx)
63 
    .await?;
64 
    sqlx::query(
65 
        "DELETE FROM terraform_state_versions WHERE repo_id = $1 AND name = $2 AND id NOT IN \
66 
         (SELECT id FROM terraform_state_versions WHERE repo_id = $1 AND name = $2 ORDER BY id DESC LIMIT $3)",
67 
    )
68 
    .bind(repo_id)
69 
    .bind(name)
70 
    .bind(keep)
71 
    .execute(&mut *tx)
72 
    .await?;
73 
    tx.commit().await
74 
}
75
76 
/// Delete a state: its body goes (its versions stay, for a mistake).
77 
pub async fn delete(pool: &Pool, repo_id: i64, name: &str) -> sqlx::Result<()> {
78 
    sqlx::query("DELETE FROM terraform_states WHERE repo_id = $1 AND name = $2").bind(repo_id).bind(name).execute(pool).await?;
79 
    Ok(())
80 
}
81
82 
/// Take the lock; false when someone holds it.
83 
pub async fn lock(pool: &Pool, repo_id: i64, name: &str, id: &str, info: &str, by: &str) -> sqlx::Result<bool> {
84 
    let now = now_ms();
85 
    let r = sqlx::query(
86 
        "INSERT INTO terraform_states (repo_id, name, body, lock_id, lock_info, locked_by, locked_ms, updated_ms) \
87 
         VALUES ($1, $2, $3, $4, $5, $6, $7, $7) \
88 
         ON CONFLICT (repo_id, name) DO UPDATE SET lock_id = $4, lock_info = $5, locked_by = $6, locked_ms = $7 \
89 
         WHERE terraform_states.lock_id = ''",
90 
    )
91 
    .bind(repo_id)
92 
    .bind(name)
93 
    .bind(Vec::<u8>::new())
94 
    .bind(id)
95 
    .bind(info)
96 
    .bind(by)
97 
    .bind(now)
98 
    .execute(pool)
99 
    .await?;
100 
    Ok(r.rows_affected() == 1)
101 
}
102
103 
pub async fn unlock(pool: &Pool, repo_id: i64, name: &str) -> sqlx::Result<()> {
104 
    sqlx::query(
105 
        "UPDATE terraform_states SET lock_id = '', lock_info = '', locked_by = '', locked_ms = NULL WHERE repo_id = $1 AND name = $2",
106 
    )
107 
    .bind(repo_id)
108 
    .bind(name)
109 
    .execute(pool)
110 
    .await?;
111 
    Ok(())
112 
}