use actix_web::http::StatusCode;
use actix_web::web::Data;
use chrono::Utc;
use serde_json::{Map, Value, json};
use std::time::Duration as StdDuration;
use std::time::Instant;
use tokio::time::Duration as TokioDuration;
use tracing::{error, info, warn};
use crate::AppState;
use crate::api::gateway::lifecycle::log_gateway_operation_result;
use crate::athena::postgres_clients::ensure_catalog_database_client_loaded;
use crate::data::events::post_event;
use crate::data::outbox::{OutboxEventInsert, insert_outbox_event_tx};
use crate::utils::request_logging::LoggedRequest;
use crate::webhooks::GatewayWebhookTrigger;
use super::backend::{InsertDriverError, postgres_insert_with_timeout};
use super::config::insert_db_timeout_ms;
use super::dedupe::{build_insert_duplicate_signature, lookup_recent_unique_violation};
use super::error::{
InsertError, WindowInsertOutcome, build_error_metadata, log_gateway_insert_success,
observe_insert_error, prefilter_unique_violation_error, trace_insert_error,
};
use super::window::WindowInsertJob;
pub(crate) fn insert_request_has_update_body(body: &Value) -> bool {
body.get("update_body").is_some()
}
pub(crate) fn should_invalidate_cache_after_insert(inserted_row: &Value) -> bool {
match inserted_row {
Value::Null => false,
Value::Object(map) => !map.is_empty(),
_ => true,
}
}
fn build_diff_resource(body: &Value, insert_body: &Value, user_id: &str) -> Value {
let mut diff_resource: Value = json!({});
if let (Some(update_body), Some(insert_body_obj)) = (
body.get("update_body").and_then(Value::as_object),
insert_body.as_object(),
) {
for (key, new_value) in update_body {
if !new_value.is_null() {
let should_include = match insert_body_obj.get(key) {
Some(existing) => existing != new_value,
None => true,
};
if should_include {
diff_resource[key] = json!({
"blame": { "user_id": user_id },
"new": new_value,
"old": null,
"time": chrono::Utc::now().timestamp(),
"void": false
});
}
}
}
}
diff_resource
}
#[allow(clippy::too_many_arguments)]
pub(crate) async fn record_insert_audit_log(
_table_name: String,
_resource_id: String,
_insert_body: Value,
_company_id: String,
_organization_id: String,
_domain: String,
_user_id: String,
_message: String,
_status: String,
_source: String,
_actor_id: String,
_action: String,
_diff_resource: Value,
_meta: Value,
_success: bool,
_path: String,
_client_name: &str,
) -> Result<Value, String> {
Ok(json!({ "audit": true }))
}
pub(crate) async fn finish_postgres_insert_success_json_with_recovery(
app_state: Data<AppState>,
job: &WindowInsertJob,
inserted_row: Value,
) -> WindowInsertOutcome {
let timeout: StdDuration = TokioDuration::from_secs(120);
match tokio::time::timeout(
timeout,
finish_postgres_insert_success_json(app_state.clone(), job, inserted_row),
)
.await
{
Ok(body) => WindowInsertOutcome::Success(body),
Err(_) => {
error!(
trace_id = %job.trace_id,
client = %job.client_name,
table = %job.table_name,
"Response serialization timeout (exceeded 120s) - audit log, webhooks, or cache invalidation likely hung"
);
app_state
.metrics_state
.record_gateway_insert_window_event("response_serialization_timeout");
let mut details: Map<String, Value> = Map::new();
details.insert("table".to_string(), Value::String(job.table_name.clone()));
details.insert("client".to_string(), Value::String(job.client_name.clone()));
details.insert(
"likely_cause".to_string(),
Value::String("audit_log_or_webhook_delay".to_string()),
);
WindowInsertOutcome::Error(InsertError::new(
StatusCode::INTERNAL_SERVER_ERROR,
"response_serialization_timeout",
"Response serialization exceeded maximum time limit (120s); check audit logging, webhooks, and cache invalidation".to_string(),
job.trace_id.clone(),
details,
))
}
}
}
pub(crate) async fn finish_postgres_insert_success_json(
app_state: Data<AppState>,
job: &WindowInsertJob,
inserted_row: Value,
) -> Value {
let response_started: Instant = Instant::now();
let diff_resource: Value = build_diff_resource(&job.body, &job.insert_body, &job.user_id);
let resource_id: String = inserted_row
.as_object()
.and_then(|object| object.get(&job.resource_id_key))
.and_then(Value::as_str)
.map(str::to_string)
.unwrap_or_else(|| job.trace_id.clone());
let audit_message: String = format!(
"User {} inserted data into [{}] at {}",
job.user_id,
job.table_name,
Utc::now().timestamp()
);
let audit_log_started: Instant = Instant::now();
let audit_log_result: Result<Value, String> = record_insert_audit_log(
job.table_name.clone(),
resource_id.clone(),
job.insert_body.clone(),
job.company_id.clone(),
job.organization_id.clone(),
"domain".to_string(),
job.user_id.clone(),
audit_message,
"success".to_string(),
"request".to_string(),
job.user_id.clone(),
format!("insert-{}", job.table_name),
diff_resource,
json!({}),
true,
"/data/insert".to_string(),
&job.client_name,
)
.await;
let audit_log_duration: StdDuration = audit_log_started.elapsed();
match &audit_log_result {
Ok(audit_log_data) => {
if job.verbose_logging {
let audit_log_message = if job.ansi_enabled {
"\u{001b}[32mAudit log inserted successfully\u{001b}[0m (ATHENA_VERBOSE_LOGGING=1)"
} else {
"Audit log inserted successfully (ATHENA_VERBOSE_LOGGING=1)"
};
info!(
trace_id = %job.trace_id,
client = %job.client_name,
table = %job.table_name,
resource_id = %resource_id,
audit_log_duration_ms = audit_log_duration.as_millis(),
audit_payload = ?audit_log_data,
"{}",
audit_log_message
);
}
if audit_log_duration.as_millis() > 5000 {
warn!(
trace_id = %job.trace_id,
client = %job.client_name,
table = %job.table_name,
audit_log_duration_ms = audit_log_duration.as_millis(),
"Slow audit log insertion (>5s)"
);
}
}
Err(err) => {
error!(
trace_id = %job.trace_id,
client = %job.client_name,
table = %job.table_name,
audit_log_duration_ms = audit_log_duration.as_millis(),
error = %err,
"Failed to insert audit log"
);
}
}
let success_json: Value = json!({
"status": "success",
"success": true,
"message": "Data inserted successfully",
"data": inserted_row,
"resource_id": resource_id,
"table_name": job.table_name,
});
if job.x_publish_event {
if let Some(company_for_event) = job.resolved_company_for_event.clone() {
let event: Value = json!({
"event": "INSERT",
"resource": job.table_name.clone(),
"inserted_by_user": job.user_id.clone(),
"company_id": company_for_event.clone(),
"insert_body": job.insert_body.clone(),
});
post_event(company_for_event, event).await;
} else {
info!(
trace_id = %job.trace_id,
"Skipping publish event because company_id is missing"
);
}
}
let logged_request: LoggedRequest = LoggedRequest {
request_id: job.logged_request_id.clone(),
client_name: job.logged_client_name.clone(),
method: job.logged_method.clone(),
path: job.logged_path.clone(),
status_code: 0,
time: 0,
};
log_gateway_operation_result(
Some(app_state.get_ref()),
&logged_request,
"insert",
Some(&job.table_name),
job.operation_start,
StatusCode::OK,
Some(json!({
"resource_id": resource_id,
"client": job.client_name,
})),
);
let mut wh_headers = Vec::new();
if !job.user_id.is_empty() {
wh_headers.push(("x-user-id".to_string(), job.user_id.clone()));
}
let trigger: GatewayWebhookTrigger = GatewayWebhookTrigger {
client_name: job.client_name.clone(),
route_key: crate::webhooks::ROUTE_GATEWAY_INSERT.to_string(),
table_name: Some(job.table_name.clone()),
request_id: Some(logged_request.request_id.clone()),
request_method: job.logged_method.clone(),
request_path: job.logged_path.clone(),
headers: wh_headers,
payload: Some(job.body.clone()),
response: Some(success_json.clone()),
};
write_insert_outbox_shadow(&app_state, &job, &trigger, &success_json).await;
crate::webhooks::spawn_gateway_webhook_dispatch(app_state.clone(), trigger);
let response_duration: StdDuration = response_started.elapsed();
if response_duration.as_millis() > 10000 {
warn!(
trace_id = %job.trace_id,
client = %job.client_name,
table = %job.table_name,
response_serialize_duration_ms = response_duration.as_millis(),
"Slow response serialization phase (>10s) - check audit log, webhooks, cache invalidation"
);
} else if response_duration.as_millis() > 5000 {
info!(
trace_id = %job.trace_id,
client = %job.client_name,
table = %job.table_name,
response_serialize_duration_ms = response_duration.as_millis(),
"Response serialization took >5s"
);
}
success_json
}
async fn write_insert_outbox_shadow(
app_state: &AppState,
job: &WindowInsertJob,
trigger: &crate::webhooks::GatewayWebhookTrigger,
response_payload: &Value,
) {
let Some(logging_client) = app_state.logging_client_name.as_ref() else {
return;
};
let Some(pool) = app_state.pg_registry.get_pool(logging_client) else {
return;
};
let headers = json!({
"client_name": job.client_name,
"request_id": job.logged_request_id,
"company_id": job.resolved_company_for_event.as_deref().unwrap_or(""),
});
let mut tx = match pool.begin().await {
Ok(t) => t,
Err(err) => {
warn!(client = %job.client_name, error = %err, "Insert outbox shadow: begin failed");
return;
}
};
if job.x_publish_event {
if let Some(company) = &job.resolved_company_for_event {
let cdc_payload: Value = json!({
"event": "INSERT",
"resource": job.table_name,
"inserted_by_user": job.user_id,
"company_id": company,
"insert_body": job.insert_body,
});
let insert = OutboxEventInsert {
aggregate_type: "gateway".into(),
aggregate_id: job.table_name.clone(),
event_type: "mutation.insert".into(),
payload: cdc_payload,
headers: headers.clone(),
available_at: None,
};
if let Err(err) = insert_outbox_event_tx(&mut tx, insert).await {
warn!(client = %job.client_name, error = %err, "Insert outbox shadow: mutation event failed");
}
}
}
let wh_payload = json!({
"route_key": trigger.route_key,
"table_name": trigger.table_name,
"request_method": trigger.request_method,
"request_path": trigger.request_path,
"request_payload": trigger.payload,
"response_payload": response_payload,
"headers": trigger.headers,
});
let wh_insert = OutboxEventInsert {
aggregate_type: "gateway".into(),
aggregate_id: job.table_name.clone(),
event_type: "webhook.trigger".into(),
payload: wh_payload,
headers,
available_at: None,
};
if let Err(err) = insert_outbox_event_tx(&mut tx, wh_insert).await {
warn!(client = %job.client_name, error = %err, "Insert outbox shadow: webhook event failed");
}
if let Err(err) = tx.commit().await {
warn!(client = %job.client_name, error = %err, "Insert outbox shadow: commit failed");
}
}
pub(crate) async fn run_postgres_insert_to_outcome(
app_state: Data<AppState>,
job: WindowInsertJob,
) -> WindowInsertOutcome {
let trace_id: String = job.trace_id.clone();
let table_name: String = job.table_name.clone();
let client_name: String = job.client_name.clone();
let user_id: String = job.user_id.clone();
let metadata_user_id: Option<&str> = job.metadata_user_id.as_deref();
let metadata_company_id: Option<&str> = job.metadata_company_id.as_deref();
let metadata_organization_id: Option<&str> = job.metadata_organization_id.as_deref();
let pool = if let Some(pool) = app_state.pg_registry.get_pool(&client_name) {
pool
} else {
match ensure_catalog_database_client_loaded(app_state.get_ref(), &client_name).await {
Ok(Some(_)) => {
let Some(pool) = app_state.pg_registry.get_pool(&client_name) else {
let mut details: Map<String, Value> = build_error_metadata(
&trace_id,
Some(&table_name),
&client_name,
metadata_user_id,
metadata_company_id,
metadata_organization_id,
);
details.insert("unknown_client".to_string(), json!(client_name.clone()));
let insert_error = InsertError::new(
StatusCode::BAD_REQUEST,
"unknown_client",
format!("Postgres client '{}' is not configured", client_name),
trace_id.clone(),
details,
);
observe_insert_error(
app_state.get_ref(),
&client_name,
Some(&table_name),
Some(&job.insert_body),
&insert_error,
);
trace_insert_error(&table_name, &client_name, &user_id, &insert_error);
return WindowInsertOutcome::Error(insert_error);
};
pool
}
Ok(None) | Err(_) => {
let mut details: Map<String, Value> = build_error_metadata(
&trace_id,
Some(&table_name),
&client_name,
metadata_user_id,
metadata_company_id,
metadata_organization_id,
);
details.insert("unknown_client".to_string(), json!(client_name.clone()));
let insert_error: InsertError = InsertError::new(
StatusCode::BAD_REQUEST,
"unknown_client",
format!("Postgres client '{}' is not configured", client_name),
trace_id.clone(),
details,
);
observe_insert_error(
app_state.get_ref(),
&client_name,
Some(&table_name),
Some(&job.insert_body),
&insert_error,
);
trace_insert_error(&table_name, &client_name, &user_id, &insert_error);
return WindowInsertOutcome::Error(insert_error);
}
}
};
if let Some(signature) =
build_insert_duplicate_signature(&client_name, &table_name, &job.insert_body)
{
let dedupe_started: Instant = Instant::now();
if let Some(cached_constraint) = lookup_recent_unique_violation(&signature) {
app_state
.metrics_state
.record_gateway_insert_window_event("gateway_recent_conflict_cache_hit");
app_state
.metrics_state
.record_gateway_insert_phase_duration(
"dedupe_check",
dedupe_started.elapsed().as_secs_f64(),
);
let insert_error: InsertError = prefilter_unique_violation_error(
trace_id.clone(),
&table_name,
&client_name,
metadata_user_id,
metadata_company_id,
metadata_organization_id,
cached_constraint.as_deref(),
"recent_unique_violation_cache_window",
);
observe_insert_error(
app_state.get_ref(),
&client_name,
Some(&table_name),
Some(&job.insert_body),
&insert_error,
);
trace_insert_error(&table_name, &client_name, &user_id, &insert_error);
return WindowInsertOutcome::Error(insert_error);
}
app_state
.metrics_state
.record_gateway_insert_window_event("gateway_recent_conflict_cache_miss");
app_state
.metrics_state
.record_gateway_insert_phase_duration(
"dedupe_check",
dedupe_started.elapsed().as_secs_f64(),
);
}
let db_insert_started: Instant = Instant::now();
let db_timeout_ms: u64 = insert_db_timeout_ms(app_state.get_ref());
let insert_result: Result<Value, InsertDriverError> =
postgres_insert_with_timeout(&pool, &table_name, &job.insert_body, db_timeout_ms).await;
app_state
.metrics_state
.record_gateway_insert_phase_duration(
"db_insert",
db_insert_started.elapsed().as_secs_f64(),
);
match insert_result {
Ok(inserted_row) => {
log_gateway_insert_success(
"athena",
&trace_id,
&client_name,
&table_name,
job.operation_start.elapsed().as_millis(),
db_insert_started.elapsed().as_millis() as u64,
);
finish_postgres_insert_success_json_with_recovery(app_state, &job, inserted_row).await
}
Err(driver_error) => {
let insert_error: InsertError = InsertError::from_driver(
driver_error,
Some(&table_name),
&client_name,
metadata_user_id,
metadata_company_id,
metadata_organization_id,
trace_id,
);
observe_insert_error(
app_state.get_ref(),
&client_name,
Some(&table_name),
Some(&job.insert_body),
&insert_error,
);
trace_insert_error(&table_name, &client_name, &user_id, &insert_error);
WindowInsertOutcome::Error(insert_error)
}
}
}