WIP: use new change set approvals table

- Add change set approvals table with integration test
- Collect approval status for a change set (only current approvals right
  now and without the validity check)
- Add new approval route

Co-authored-by: Jacob Helwig <[email protected]>
Co-authored-by: John Obelenus <[email protected]>
Signed-off-by: Nick Gerace <[email protected]>

lib/dal/src/change_set.rs

@@ -26,6 +26,7 @@ use crate::{
     WorkspaceError,
 };
 
+pub mod approval;
 pub mod event;
 pub mod status;
 pub mod view;

lib/dal/src/change_set/approval.rs

@@ -0,0 +1,113 @@
+use chrono::{DateTime, Utc};
+use serde::{Deserialize, Serialize};
+use si_data_pg::{PgError, PgRow};
+use si_id::{ChangeSetApprovalId, ChangeSetId, UserPk};
+use telemetry::prelude::*;
+use thiserror::Error;
+
+pub use si_events::ChangeSetApprovalStatus;
+
+use crate::{DalContext, HistoryActor, TransactionsError};
+
+#[derive(Debug, Error)]
+pub enum ChangeSetApprovalError {
+    #[error("invalid user for creating a change set approval")]
+    InvalidUserForCreation,
+    #[error("pg error: {0}")]
+    Pg(#[from] PgError),
+    #[error("strum parse error: {0}")]
+    StrumParse(#[from] strum::ParseError),
+    #[error("transactions error: {0}")]
+    Transactions(#[from] TransactionsError),
+}
+
+type Result<T> = std::result::Result<T, ChangeSetApprovalError>;
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct ChangeSetApproval {
+    id: ChangeSetApprovalId,
+    created_at: DateTime<Utc>,
+    updated_at: DateTime<Utc>,
+    change_set_id: ChangeSetId,
+    status: ChangeSetApprovalStatus,
+    user_id: UserPk,
+    checksum: String,
+}
+
+impl TryFrom<PgRow> for ChangeSetApproval {
+    type Error = ChangeSetApprovalError;
+
+    fn try_from(value: PgRow) -> std::result::Result<Self, Self::Error> {
+        let status_string: String = value.try_get("status")?;
+        let status = ChangeSetApprovalStatus::try_from(status_string.as_str())?;
+        Ok(Self {
+            id: value.try_get("id")?,
+            created_at: value.try_get("created_at")?,
+            updated_at: value.try_get("updated_at")?,
+            change_set_id: value.try_get("change_set_id")?,
+            status,
+            user_id: value.try_get("user_id")?,
+            checksum: value.try_get("checksum")?,
+        })
+    }
+}
+
+impl ChangeSetApproval {
+    #[instrument(name = "change_set.approval.new", level = "info", skip_all)]
+    pub async fn new(
+        ctx: &DalContext,
+        status: ChangeSetApprovalStatus,
+        checksum: String,
+    ) -> Result<Self> {
+        let change_set_id = ctx.change_set_id();
+        let user_id = match ctx.history_actor() {
+            HistoryActor::User(user_id) => user_id,
+            HistoryActor::SystemInit => return Err(ChangeSetApprovalError::InvalidUserForCreation),
+        };
+        let row = ctx
+            .txns()
+            .await?
+            .pg()
+            .query_one(
+                "INSERT INTO change_set_approvals (change_set_id, status, user_id, checksum) VALUES ($1, $2, $3, $4) RETURNING *",
+                &[&change_set_id, &status, &user_id, &checksum]
+            )
+            .await?;
+        Self::try_from(row)
+    }
+
+    pub fn id(&self) -> ChangeSetApprovalId {
+        self.id
+    }
+
+    pub fn status(&self) -> ChangeSetApprovalStatus {
+        self.status
+    }
+
+    pub fn user_id(&self) -> UserPk {
+        self.user_id
+    }
+
+    pub fn checksum(&self) -> &str {
+        self.checksum.as_str()
+    }
+
+    #[instrument(name = "change_set.approval.list", level = "info", skip_all)]
+    pub async fn list(ctx: &DalContext) -> Result<Vec<Self>> {
+        let change_set_id = ctx.change_set_id();
+        let rows = ctx
+            .txns()
+            .await?
+            .pg()
+            .query(
+                "SELECT * from change_set_approvals WHERE change_set_id = $1 ORDER BY id ASC",
+                &[&change_set_id],
+            )
+            .await?;
+        let mut approvals = Vec::with_capacity(rows.len());
+        for row in rows {
+            approvals.push(Self::try_from(row)?);
+        }
+        Ok(approvals)
+    }
+}

lib/dal/src/migrations/U3500__change_set_approvals.sql

@@ -0,0 +1,10 @@
+CREATE change_set_approvals
+(
+    id                         ident primary key        NOT NULL DEFAULT ident_create_v1(),
+    created_at                 timestamp with time zone NOT NULL DEFAULT CLOCK_TIMESTAMP(),
+    updated_at                 timestamp with time zone NOT NULL DEFAULT CLOCK_TIMESTAMP(),
+    change_set_id ident NOT NULL,
+    status text NOT NULL,
+    user_id ident NOT NULL,
+    checksum text NOT NULL
+);

lib/dal/tests/integration_test/change_set.rs

@@ -12,6 +12,8 @@ use itertools::Itertools;
 use pretty_assertions_sorted::assert_eq;
 use std::collections::HashSet;
 
+mod approval;
+
 #[test]
 async fn open_change_sets(ctx: &mut DalContext) {
     let view = OpenChangeSetsView::assemble(ctx)

lib/dal/tests/integration_test/change_set/approval.rs

@@ -0,0 +1,32 @@
+use dal::change_set::approval::{ChangeSetApproval, ChangeSetApprovalStatus};
+use dal::DalContext;
+use dal_test::color_eyre::eyre::OptionExt;
+use dal_test::{test, Result};
+use pretty_assertions_sorted::assert_eq;
+
+#[test]
+async fn new(ctx: &mut DalContext) -> Result<()> {
+    let status = ChangeSetApprovalStatus::Approved;
+    // FIXME(nick): use a real checksum here.
+    let checksum = "FIXME".to_string();
+
+    let new_approval = ChangeSetApproval::new(ctx, status, checksum).await?;
+    assert_eq!(
+        status,                // expectd
+        new_approval.status()  // actual
+    );
+
+    let mut approvals = ChangeSetApproval::list(ctx).await?;
+    let approval = approvals.pop().ok_or_eyre("unexpected empty approvals")?;
+    assert!(approvals.is_empty());
+    assert_eq!(
+        new_approval.status(), // expected
+        approval.status()      // actual
+    );
+    assert_eq!(
+        new_approval.id(), // expected
+        approval.id()      // actual
+    );
+
+    Ok(())
+}

lib/sdf-server/src/service/v2/change_set.rs

@@ -27,9 +27,15 @@ mod rename;
 mod reopen;
 mod request_approval;
 
+// NOTE(nick): move these to the above group and remove old modules once the feature flag has been removed;
+mod approval_status;
+mod approve_v2;
+
 #[remain::sorted]
 #[derive(Debug, Error)]
 pub enum Error {
+    #[error("change set approval error: {0}")]
+    Approval(#[from] dal::change_set::approval::ChangeSetApprovalError),
     #[error("change set error: {0}")]
     ChangeSet(#[from] dal::ChangeSetError),
     #[error("change set apply error: {0}")]
@@ -158,7 +164,9 @@ pub fn v2_routes(state: AppState) -> Router<AppState> {
                         permissions::Permission::Approve,
                     )),
                 )
-                .route("/rename", post(rename::rename)),
+                .route("/rename", post(rename::rename))
+                .route("/approval_status", get(approval_status::approval_status))
+                .route("/approve_v2", post(approve_v2::approve)),
         )
         .route("/", get(list::list_actionable))
 }

lib/sdf-server/src/service/v2/change_set/approval_status.rs

@@ -0,0 +1,38 @@
+use axum::{
+    extract::{Host, OriginalUri, Path, State},
+    Json,
+};
+use dal::{change_set::approval::ChangeSetApproval, ChangeSet, ChangeSetId, WorkspacePk};
+
+use super::{AppState, Error, Result};
+use crate::{
+    extract::{AccessBuilder, HandlerContext, PosthogClient},
+    track,
+};
+
+pub async fn approval_status(
+    HandlerContext(builder): HandlerContext,
+    AccessBuilder(access_builder): AccessBuilder,
+    Path((_workspace_pk, change_set_id)): Path<(WorkspacePk, ChangeSetId)>,
+) -> Result<Json<si_frontend_types::ChangeSetApprovals>> {
+    let ctx = builder
+        .build(access_builder.build(change_set_id.into()))
+        .await?;
+
+    let approvals = ChangeSetApproval::list(&ctx).await?;
+    let mut current = Vec::with_capacity(approvals.len());
+    for approval in approvals {
+        current.push(si_frontend_types::ChangeSetApproval {
+            user_id: approval.user_id(),
+            status: approval.status(),
+            // FIXME(nick): make this real!
+            is_valid: true,
+        })
+    }
+
+    Ok(Json(si_frontend_types::ChangeSetApprovals {
+        // FIXME(nick): get requirements.
+        required: Vec::new(),
+        current,
+    }))
+}

lib/sdf-server/src/service/v2/change_set/approve_v2.rs

@@ -0,0 +1,77 @@
+use axum::{
+    extract::{Host, OriginalUri, Path},
+    Json,
+};
+use dal::{change_set::approval::ChangeSetApproval, ChangeSet, ChangeSetId, WorkspacePk, WsEvent};
+use serde::Deserialize;
+use si_events::{audit_log::AuditLogKind, ChangeSetApprovalStatus};
+
+use super::{post_to_webhook, Error, Result};
+use crate::{
+    extract::{AccessBuilder, HandlerContext, PosthogClient},
+    track,
+};
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct Request {
+    pub status: ChangeSetApprovalStatus,
+}
+
+pub async fn approve(
+    HandlerContext(builder): HandlerContext,
+    AccessBuilder(request_ctx): AccessBuilder,
+    PosthogClient(posthog_client): PosthogClient,
+    OriginalUri(original_uri): OriginalUri,
+    Host(host_name): Host,
+    Path((workspace_pk, change_set_id)): Path<(WorkspacePk, ChangeSetId)>,
+    Json(request): Json<Request>,
+) -> Result<()> {
+    let ctx = builder
+        .build(request_ctx.build(change_set_id.into()))
+        .await?;
+
+    // Ensure that DVU roots are empty before continuing?
+    // todo(brit): maybe we can get away without this. Ex: Approve a PR before tests finish
+    if !ctx
+        .workspace_snapshot()?
+        .get_dependent_value_roots()
+        .await?
+        .is_empty()
+    {
+        // TODO(nick): we should consider requiring this check in integration tests too. Why did I
+        // not do this at the time of writing? Tests have multiple ways to call "apply", whether
+        // its via helpers or through the change set methods directly. In addition, they test
+        // for success and failure, not solely for success. We should still do this, but not within
+        // the PR corresponding to when this message was written.
+        return Err(Error::DvuRootsNotEmpty(ctx.change_set_id()));
+    }
+
+    let change_set = ChangeSet::find(&ctx, ctx.visibility().change_set_id)
+        .await?
+        .ok_or(Error::ChangeSetNotFound(ctx.change_set_id()))?;
+    // FIXME(nick): put the real checksum here.
+    ChangeSetApproval::new(&ctx, request.status, "FIXME".to_string()).await?;
+
+    track(
+        &posthog_client,
+        &ctx,
+        &original_uri,
+        &host_name,
+        "approve_change_set_apply",
+        serde_json::json!({
+            "merged_change_set": change_set.id,
+        }),
+    );
+    ctx.write_audit_log(
+        AuditLogKind::ApproveChangeSetApply {
+            from_status: change_set.status.into(),
+        },
+        change_set.name,
+    )
+    .await?;
+
+    ctx.commit().await?;
+
+    Ok(())
+}

lib/si-events-rs/src/change_set_approval.rs

@@ -0,0 +1,19 @@
+use postgres_types::ToSql;
+use serde::{Deserialize, Serialize};
+use strum::{AsRefStr, Display, EnumString};
+
+#[remain::sorted]
+#[derive(
+    AsRefStr, Deserialize, Serialize, Debug, Display, EnumString, PartialEq, Eq, Copy, Clone, ToSql,
+)]
+pub enum ChangeSetApprovalStatus {
+    Approved,
+}
+
+#[remain::sorted]
+#[derive(Debug, Clone, Deserialize, Serialize)]
+pub enum ChangeSetApprovalKind {
+    Func,
+    SchemaVariant,
+    View,
+}

lib/si-events-rs/src/lib.rs

@@ -10,6 +10,7 @@ pub use si_id::ulid;
 
 mod actor;
 mod cas;
+mod change_set_approval;
 mod change_set_status;
 mod event_session;
 mod func;
@@ -30,6 +31,7 @@ pub use crate::{
     actor::Actor,
     actor::UserPk,
     cas::CasValue,
+    change_set_approval::{ChangeSetApprovalKind, ChangeSetApprovalStatus},
     change_set_status::ChangeSetStatus,
     content_hash::ContentHash,
     encrypted_secret::EncryptedSecretKey,

lib/si-frontend-types-rs/src/change_set.rs

@@ -1,6 +1,9 @@
 use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize};
-use si_events::{ChangeSetId, ChangeSetStatus};
+use si_events::{
+    ulid::Ulid, ChangeSetApprovalKind, ChangeSetApprovalStatus, ChangeSetId, ChangeSetStatus,
+    UserPk,
+};
 
 #[derive(Clone, Debug, Deserialize, Serialize, Eq, PartialEq)]
 #[serde(rename_all = "camelCase")]
@@ -19,3 +22,33 @@ pub struct ChangeSet {
     pub reviewed_by_user: Option<String>,
     pub reviewed_at: Option<DateTime<Utc>>,
 }
+
+#[derive(Debug, Clone, Deserialize, Serialize)]
+pub struct ChangeSetApprovals {
+    pub required: Vec<ChangeSetRequiredApproval>,
+    pub current: Vec<ChangeSetApproval>,
+}
+
+#[derive(Debug, Clone, Deserialize, Serialize)]
+pub struct ChangeSetRequiredApproval {
+    // What is the kind of the entity corresponding to the ID?
+    kind: ChangeSetApprovalKind,
+    // What is the ID of the entity that is requiring approvals?
+    id: Ulid,
+    // What is the minimum number needed?
+    number: usize,
+    // Is it satisfied?
+    is_satisfied: bool,
+    // Who can satisfy this?
+    users: Vec<UserPk>,
+}
+
+#[derive(Debug, Clone, Deserialize, Serialize)]
+pub struct ChangeSetApproval {
+    // Who approved this?
+    pub user_id: UserPk,
+    // What kind of approval did they do (including negative)?
+    pub status: ChangeSetApprovalStatus,
+    // Is this still valid?
+    pub is_valid: bool,
+}