//! This module provides audit logging functionality to the rest of the crate.
use audit_database::{
AuditDatabaseContext,
AuditDatabaseError,
AuditLogRow,
};
use audit_logs_stream::AuditLogsStreamError;
use pending_events::PendingEventsError;
use serde::{
Deserialize,
Serialize,
};
use shuttle_server::{
Shuttle,
ShuttleError,
};
use si_events::audit_log::{
AuditLog,
AuditLogKind,
};
use si_id::{
ChangeSetId,
WorkspacePk,
};
use telemetry::prelude::*;
use telemetry_utils::monotonic;
use thiserror::Error;
use tokio_util::task::TaskTracker;
use crate::{
ChangeSet,
ChangeSetError,
ChangeSetStatus,
DalContext,
TransactionsError,
WsEvent,
WsEventResult,
WsPayload,
};
#[remain::sorted]
#[derive(Debug, Error)]
pub enum AuditLoggingError {
#[error("audit database error: {0}")]
AuditDatabase(#[from] AuditDatabaseError),
#[error("audit logs stream error: {0}")]
AuditLogsStream(#[from] AuditLogsStreamError),
#[error("change set error: {0}")]
ChangeSet(#[from] Box<ChangeSetError>),
#[error("pending events error: {0}")]
PendingEventsError(#[from] PendingEventsError),
#[error("shuttle error: {0}")]
Shuttle(#[from] ShuttleError),
#[error("si db error: {0}")]
SiDb(#[from] si_db::Error),
#[error("transactions error: {0}")]
Transactions(#[from] Box<TransactionsError>),
}
type Result<T> = std::result::Result<T, AuditLoggingError>;
/// Publishes all pending [`AuditLogs`](AuditLog) to the audit logs stream for the event session.
///
/// Provide the "override" [`EventSessionId`] if you'd like to use a different identifier than
/// the one on [`self`](DalContext).
///
/// _Warning: the subject for the event session must have a [final message](write_final_message)._
#[instrument(
name = "audit_logging.publish_pending",
level = "debug",
skip_all,
fields(override_event_session_id)
)]
pub(crate) async fn publish_pending(
ctx: &DalContext,
tracker: Option<TaskTracker>,
override_event_session_id: Option<si_events::EventSessionId>,
) -> Result<()> {
// TODO(nick): nuke this from intergalactic orbit. Then do it again.
let workspace_id = match ctx.workspace_pk() {
Ok(workspace_id) => workspace_id,
Err(TransactionsError::SiDb(si_db_err))
if matches!(si_db_err.as_ref(), si_db::Error::NoWorkspace) =>
{
return Ok(());
}
Err(err) => return Err(AuditLoggingError::Transactions(Box::new(err))),
};
let (tracker, provided_tracker) = match tracker {
Some(provided_tracker) => (provided_tracker, false),
None => (TaskTracker::new(), true),
};
// Emit metric for pending events published during this event session
let count = ctx
.pending_audit_logs_count()
.load(std::sync::atomic::Ordering::Relaxed);
monotonic!(pending_events_published = count);
// Get a handle on the source and destination streams from the cached instances.
let source_stream = ctx.jetstream_streams().pending_events();
let destination_stream = ctx.jetstream_streams().audit_logs();
// Create a shuttle instance for shuttling audit logs from the pending events stream.
let audit_logs_shuttle = Shuttle::new(
ctx.nats_conn().to_owned(),
source_stream.stream().await?,
source_stream.subject_for_audit_log(
workspace_id,
ctx.change_set_id(),
match override_event_session_id {
Some(override_id) => override_id,
None => ctx.event_session_id(),
},
),
destination_stream.publishing_subject_for_workspace(workspace_id),
)
.await?;
// Run the audit logs shuttle instance. If a tracker has been provided, we can spawn the
// shuttle instance using it. If we are using a tracker purely within this function, we cannot
// reliably use it to run the shuttle instance, so we will close and wait once shuttle exits.
let ctx_clone_for_ws_event = ctx.clone();
if provided_tracker {
tracker.spawn(async move {
if let Err(err) = audit_logs_shuttle.try_run().await {
error!(?err, "audit logs shuttle error");
}
match ChangeSet::find(
&ctx_clone_for_ws_event,
ctx_clone_for_ws_event.change_set_id(),
)
.await
{
Ok(Some(change_set)) => {
match WsEvent::audit_logs_published(
&ctx_clone_for_ws_event,
change_set.id,
change_set.status,
)
.await
{
Ok(event) => {
if let Err(err) =
event.publish_immediately(&ctx_clone_for_ws_event).await
{
error!(?err, "error when publishing ws event for audit logs");
}
}
Err(err) => error!(?err, "error when creating ws event for audit logs"),
}
}
Ok(None) => {
trace!("skipping ws event creation for audit logs: no change set found")
}
Err(err) => error!(
?err,
"error when attempting to find change set for ws event for audit logs"
),
}
});
} else {
// TODO(nick): this needs a tracker. In fact, func runner does too. We'll need a long term
// solution for spwaning tasks in the dal.
tokio::spawn(async move {
if let Err(err) = audit_logs_shuttle.try_run().await {
error!(?err, "audit logs shuttle error");
}
tracker.close();
tracker.wait().await;
match ChangeSet::find(
&ctx_clone_for_ws_event,
ctx_clone_for_ws_event.change_set_id(),
)
.await
{
Ok(Some(change_set)) => {
match WsEvent::audit_logs_published(
&ctx_clone_for_ws_event,
change_set.id,
change_set.status,
)
.await
{
Ok(event) => {
if let Err(err) =
event.publish_immediately(&ctx_clone_for_ws_event).await
{
error!(?err, "error when publishing ws event for audit logs");
}
}
Err(err) => error!(?err, "error when creating ws event for audit logs"),
}
}
Ok(None) => {
trace!("skipping ws event creation for audit logs: no change set found")
}
Err(err) => error!(
?err,
"error when attempting to find change set for ws event for audit logs"
),
}
});
}
Ok(())
}
#[instrument(
name = "audit_logging.write",
level = "debug",
skip_all,
fields(kind, entity_name, override_destination_change_set_id)
)]
pub(crate) async fn write(
ctx: &DalContext,
kind: AuditLogKind,
entity_name: String,
override_destination_change_set_id: Option<si_events::ChangeSetId>,
) -> Result<()> {
// TODO(nick): nuke this from intergalactic orbit. Then do it again.
let workspace_id = match ctx.workspace_pk() {
Ok(workspace_id) => workspace_id,
Err(TransactionsError::SiDb(si_db_err))
if matches!(si_db_err.as_ref(), si_db::Error::NoWorkspace) =>
{
return Ok(());
}
Err(err) => return Err(AuditLoggingError::Transactions(Box::new(err))),
};
let destination_change_set_id =
override_destination_change_set_id.unwrap_or(ctx.change_set_id());
let pending_events_stream = ctx.jetstream_streams().pending_events();
pending_events_stream
.publish_audit_log(
workspace_id,
ctx.change_set_id(),
ctx.event_session_id(),
&AuditLog::new(
ctx.events_actor(),
kind,
entity_name,
destination_change_set_id,
ctx.authentication_method(),
),
destination_change_set_id,
)
.await?;
// Increment counter for metrics
ctx.pending_audit_logs_count()
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
Ok(())
}
#[instrument(name = "audit_logging.write_final_message", level = "debug", skip_all)]
pub(crate) async fn write_final_message(ctx: &DalContext) -> Result<()> {
// TODO(nick): nuke this from intergalactic orbit. Then do it again.
let workspace_id = match ctx.workspace_pk() {
Ok(workspace_id) => workspace_id,
Err(TransactionsError::SiDb(si_db_err))
if matches!(si_db_err.as_ref(), si_db::Error::NoWorkspace) =>
{
return Ok(());
}
Err(err) => return Err(AuditLoggingError::Transactions(Box::new(err))),
};
let pending_events_stream = ctx.jetstream_streams().pending_events();
pending_events_stream
.publish_audit_log_final_message(workspace_id, ctx.change_set_id(), ctx.event_session_id())
.await?;
Ok(())
}
#[instrument(
name = "audit_logging.list",
level = "debug",
skip_all,
fields(size, sort_ascending)
)]
pub async fn list(
ctx: &DalContext,
audit_database_context: &AuditDatabaseContext,
size: usize,
sort_ascending: bool,
) -> Result<(Vec<AuditLogRow>, bool)> {
let (workspace_id, change_set_ids) = prepare_accessor_query(ctx).await?;
Ok(AuditLogRow::list(
audit_database_context,
workspace_id,
change_set_ids,
size,
sort_ascending,
)
.await?)
}
#[instrument(
name = "audit_logging.list_for_component",
level = "debug",
skip_all,
fields(size, sort_ascending, component_id)
)]
pub async fn list_for_component(
ctx: &DalContext,
audit_database_context: &AuditDatabaseContext,
component_id: crate::ComponentId,
size: usize,
sort_ascending: bool,
) -> Result<(Vec<AuditLogRow>, bool)> {
let (workspace_id, change_set_ids) = prepare_accessor_query(ctx).await?;
Ok(AuditLogRow::list_for_component(
audit_database_context,
workspace_id,
change_set_ids,
component_id,
size,
sort_ascending,
)
.await?)
}
async fn prepare_accessor_query(ctx: &DalContext) -> Result<(WorkspacePk, Vec<ChangeSetId>)> {
let workspace_id = ctx.workspace_pk().map_err(Box::new)?;
let change_set_id = ctx.change_set_id();
let change_set_ids = {
let mut change_set_ids = vec![change_set_id];
if ctx
.get_workspace_default_change_set_id()
.await
.map_err(Box::new)?
== change_set_id
{
// NOTE(nick,fletcher,brit,paul): we need to decide what this entails on HEAD in the long term. For now,
// it is all non-open, non-abandoned change sets... which are just the applied ones. In the future, we may
// or will need to ability to tell a story about abandoned change sets. This is for future us or future
// victims to solve. Good luck!
for applied_change_set in ChangeSet::list_all_applied(ctx, workspace_id)
.await
.map_err(Box::new)?
{
change_set_ids.push(applied_change_set.id);
}
}
change_set_ids
};
Ok((workspace_id, change_set_ids))
}
#[derive(Clone, Deserialize, Serialize, Debug, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct AuditLogsPublishedPayload {
change_set_id: crate::ChangeSetId,
change_set_status: ChangeSetStatus,
}
impl WsEvent {
pub async fn audit_logs_published(
ctx: &DalContext,
change_set_id: crate::ChangeSetId,
change_set_status: ChangeSetStatus,
) -> WsEventResult<Self> {
WsEvent::new(
ctx,
WsPayload::AuditLogsPublished(AuditLogsPublishedPayload {
change_set_id,
change_set_status,
}),
)
.await
}
}
