From 9ae18b071ee9efd9803c9cd0b167112be231d8d1 Mon Sep 17 00:00:00 2001 From: max funk Date: Tue, 3 Mar 2026 07:38:13 -0800 Subject: [PATCH 01/12] remove global env var --- infra/terraform/aws/modules/ci/v001/integ.tf | 15 ++++++--------- 1 file changed, 6 insertions(+), 9 deletions(-) diff --git a/infra/terraform/aws/modules/ci/v001/integ.tf b/infra/terraform/aws/modules/ci/v001/integ.tf index 107174c80..95bdd0757 100644 --- a/infra/terraform/aws/modules/ci/v001/integ.tf +++ b/infra/terraform/aws/modules/ci/v001/integ.tf @@ -19,10 +19,6 @@ resource "aws_codebuild_project" "integ" { name = "ENV_ID" value = var.env_id } - environment_variable { - name = "ENV" - value = var.env - } environment_variable { name = "AWS_ACCOUNT_ID" value = data.aws_caller_identity.current.account_id @@ -55,11 +51,11 @@ resource "aws_codebuild_project" "integ" { - npx playwright install-deps # pull pre-built service images from ecr and tag as latest for docker compose - aws ecr get-login-password --region $REGION | docker login --username AWS --password-stdin $AWS_ACCOUNT_ID.dkr.ecr.$REGION.amazonaws.com - - export ECR_URI=$AWS_ACCOUNT_ID.dkr.ecr.$REGION.amazonaws.com/$ENV_ID/$ENV + - export ECR_URI=$AWS_ACCOUNT_ID.dkr.ecr.$REGION.amazonaws.com/$ENV_ID/${var.env} - | for SVC in ${join(" ", var.service_names)}; do echo "pulling $SVC..." - TAG=$(aws ecr describe-images --repository-name $ENV_ID/$ENV/$SVC --region $REGION --query 'sort_by(imageDetails,&imagePushedAt)[-1].imageTags[0]' --output text 2>/dev/null || echo "") + TAG=$(aws ecr describe-images --repository-name $ENV_ID/${var.env}/$SVC --region $REGION --query 'sort_by(imageDetails,&imagePushedAt)[-1].imageTags[0]' --output text 2>/dev/null || echo "") if [ -n "$TAG" ] && [ "$TAG" != "None" ]; then docker pull $ECR_URI/$SVC:$TAG docker tag $ECR_URI/$SVC:$TAG $SVC:latest @@ -69,13 +65,14 @@ resource "aws_codebuild_project" "integ" { done build: commands: - # build storage (postgres), then start all services with pre-built images + # build postgres base stage for cache_from, then build full image + - docker build --target base -t mxf-postgres-base -f docker/bitnami-postgres.Dockerfile . - docker compose -f docker/storage.yaml build - docker compose -f docker/storage.yaml -f docker/services.yaml up -d --no-build - until docker exec mxf-postgres-1 pg_isready -U postgres; do sleep 1; done - # run tests + # run tests (ENV must be unset so test-reset.sh uses local defaults) - make --no-print-directory -C crates/pg test-db - - make --no-print-directory -C crates/redisclient test-cache + - make --no-print-directory -C crates/cache test-cache - make --no-print-directory -C tests test-local - make --no-print-directory -C client test post_build: From 8e2def8b7b577975fbbb26e2f3c78fd4388a1bdb Mon Sep 17 00:00:00 2001 From: max funk Date: Tue, 3 Mar 2026 07:43:02 -0800 Subject: [PATCH 02/12] add envvar crate --- Cargo.toml | 1 + crates/envvar/Cargo.toml | 6 ++++++ crates/envvar/src/lib.rs | 33 +++++++++++++++++++++++++++++++++ 3 files changed, 40 insertions(+) create mode 100644 crates/envvar/Cargo.toml create mode 100644 crates/envvar/src/lib.rs diff --git a/Cargo.toml b/Cargo.toml index c199a1301..5d748fd1b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,6 +4,7 @@ resolver = "2" members = [ "crates/cache", "crates/cognitoidp", + "crates/envvar", "crates/httpclient", "crates/pg", "crates/pubsub", diff --git a/crates/envvar/Cargo.toml b/crates/envvar/Cargo.toml new file mode 100644 index 000000000..67e4bc96f --- /dev/null +++ b/crates/envvar/Cargo.toml @@ -0,0 +1,6 @@ +[package] +name = "envvar" +version = "0.1.0" +edition = "2021" +rust-version.workspace = true +description = "shared env var reading" diff --git a/crates/envvar/src/lib.rs b/crates/envvar/src/lib.rs new file mode 100644 index 000000000..a2e62e709 --- /dev/null +++ b/crates/envvar/src/lib.rs @@ -0,0 +1,33 @@ +use std::env; + +pub fn required(var: &str) -> Result { + env::var(var).map_err(|e| format!("{var} not set: {e}")) +} + +pub fn optional(var: &str, default: &str) -> String { + env::var(var).unwrap_or_else(|_| default.to_string()) +} + +pub fn redis_uri() -> Result { + let db = required("REDIS_DB")?; + let host = required("REDIS_HOST")?; + let port = required("REDIS_PORT")?; + let username = required("REDIS_USERNAME")?; + let password = required("REDIS_PASSWORD")?; + Ok(format!("redis://{username}:{password}@{host}:{port}/{db}")) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn it_formats_error_with_var_name() { + let result = required("NONEXISTENT_TEST_VAR"); + let err = result.unwrap_err(); + assert_eq!( + err, + "NONEXISTENT_TEST_VAR not set: environment variable not found" + ); + } +} From 0f42b381eca6da3b421365698e273a4952ef4a70 Mon Sep 17 00:00:00 2001 From: max funk Date: Tue, 3 Mar 2026 07:49:12 -0800 Subject: [PATCH 03/12] use new envvar crate --- crates/cache/Cargo.toml | 1 + crates/cache/src/lib.rs | 14 ++++- crates/cache/src/redis.rs | 61 +++++-------------- crates/pubsub/Cargo.toml | 1 + crates/pubsub/src/redis.rs | 7 +-- crates/queue/Cargo.toml | 1 + crates/queue/src/redis.rs | 7 +-- crates/uribuilder/Cargo.toml | 1 + crates/uribuilder/src/lib.rs | 2 +- services/auto-confirm/Cargo.toml | 1 + services/auto-confirm/src/main.rs | 4 +- services/auto-transact/Cargo.toml | 1 + services/balance-by-account/Cargo.toml | 1 + services/event/src/events/cron.rs | 2 +- services/event/src/events/gdp.rs | 2 +- services/event/src/events/threshold_profit.rs | 2 +- services/graphql/Cargo.toml | 1 + services/measure/Cargo.toml | 1 + services/request-approve/Cargo.toml | 1 + services/request-by-id/Cargo.toml | 1 + services/request-create/Cargo.toml | 1 + services/requests-by-account/Cargo.toml | 1 + services/rule/Cargo.toml | 1 + services/transaction-by-id/Cargo.toml | 1 + services/transactions-by-account/Cargo.toml | 1 + 25 files changed, 51 insertions(+), 66 deletions(-) diff --git a/crates/cache/Cargo.toml b/crates/cache/Cargo.toml index f010c945c..2c1664a06 100644 --- a/crates/cache/Cargo.toml +++ b/crates/cache/Cargo.toml @@ -14,6 +14,7 @@ thiserror = "2.0" serde_json = "1.0" tracing = "0.1" tokio = { version = "1.49", features = ["macros", "rt-multi-thread"] } +envvar = { path = "../envvar" } types = { path = "../types" } # redis diff --git a/crates/cache/src/lib.rs b/crates/cache/src/lib.rs index 0e146ecd7..019d4af10 100644 --- a/crates/cache/src/lib.rs +++ b/crates/cache/src/lib.rs @@ -25,6 +25,12 @@ pub enum CacheError { DeserializationError(String), } +impl From for CacheError { + fn from(s: String) -> Self { + CacheError::ConnectionError(s) + } +} + #[async_trait] pub trait Cache: Send + Sync { // transaction item rules by state @@ -123,7 +129,13 @@ pub async fn new() -> Option> { } } } else if env::var("REDIS_HOST").is_ok() { - let client = RedisClient::new().await; + let client = match RedisClient::new().await { + Ok(c) => c, + Err(e) => { + tracing::warn!("redis client creation failed: {}", e); + return None; + } + }; if let Err(e) = client.init().await { tracing::warn!("redis init failed: {}", e); None diff --git a/crates/cache/src/redis.rs b/crates/cache/src/redis.rs index 8d87a172f..70999470f 100644 --- a/crates/cache/src/redis.rs +++ b/crates/cache/src/redis.rs @@ -15,28 +15,20 @@ pub struct RedisClient { } impl RedisClient { - pub async fn new() -> Self { - let redis_uri = Self::redis_uri_from_env(); - let redis_config = Config::from_url(&redis_uri).unwrap(); - let redis_client = Builder::from_config(redis_config).build().unwrap(); - Self { + pub async fn new() -> Result { + let redis_uri = Self::redis_uri_from_env()?; + let redis_config = + Config::from_url(&redis_uri).map_err(|e| CacheError::ConnectionError(e.to_string()))?; + let redis_client = Builder::from_config(redis_config) + .build() + .map_err(|e| CacheError::ConnectionError(e.to_string()))?; + Ok(Self { inner: redis_client, - } + }) } - fn redis_uri_from_env() -> String { - let redis_db = std::env::var("REDIS_DB").unwrap(); - let redis_host = std::env::var("REDIS_HOST").unwrap(); - let redis_port = std::env::var("REDIS_PORT").unwrap(); - let redis_username = std::env::var("REDIS_USERNAME").unwrap(); - let redis_password = std::env::var("REDIS_PASSWORD").unwrap(); - redis_uri( - &redis_db, - &redis_host, - &redis_port, - &redis_username, - &redis_password, - ) + fn redis_uri_from_env() -> Result { + Ok(envvar::redis_uri()?) } pub async fn init(&self) -> Result<(), Error> { @@ -111,16 +103,6 @@ impl RedisClient { } } -fn redis_uri( - redis_db: &str, - redis_host: &str, - redis_port: &str, - redis_username: &str, - redis_password: &str, -) -> String { - format!("redis://{redis_username}:{redis_password}@{redis_host}:{redis_port}/{redis_db}") -} - #[async_trait] impl Cache for RedisClient { async fn get_transaction_item_rules( @@ -387,23 +369,6 @@ end mod tests { use super::*; - #[test] - fn it_creates_a_redis_conn_uri() { - let redis_db = "0"; - let redis_host = "localhost"; - let redis_port = "6379"; - let redis_username = "admin"; - let redis_password = "password"; - let uri = redis_uri( - redis_db, - redis_host, - redis_port, - redis_username, - redis_password, - ); - assert_eq!(uri, "redis://admin:password@localhost:6379/0"); - } - #[test] fn it_creates_rules_state_key() { let key = CacheKey::rules_state(AccountRole::Creditor, "California").to_string(); @@ -448,7 +413,9 @@ mod integration_tests { use types::rule::{ApprovalRuleInstance, TransactionItemRuleInstance}; async fn get_client() -> RedisClient { - let client = RedisClient::new().await; + let client = RedisClient::new() + .await + .expect("failed to create redis client"); client.init().await.expect("failed to init redis"); client } diff --git a/crates/pubsub/Cargo.toml b/crates/pubsub/Cargo.toml index e3dacc8d9..da2df5c87 100644 --- a/crates/pubsub/Cargo.toml +++ b/crates/pubsub/Cargo.toml @@ -7,6 +7,7 @@ description = "pubsub trait with redis and sns implementations" [dependencies] async-trait = "0.1.73" +envvar = { path = "../envvar" } thiserror = "2.0" tracing = "0.1" tokio = { version = "1.49", features = ["macros", "rt-multi-thread", "sync"] } diff --git a/crates/pubsub/src/redis.rs b/crates/pubsub/src/redis.rs index 0c3c35ac2..87e96b126 100644 --- a/crates/pubsub/src/redis.rs +++ b/crates/pubsub/src/redis.rs @@ -46,12 +46,7 @@ impl RedisPubSub { } fn redis_uri_from_env() -> String { - let redis_db = std::env::var("REDIS_DB").unwrap_or_else(|_| "0".to_string()); - let redis_host = std::env::var("REDIS_HOST").unwrap(); - let redis_port = std::env::var("REDIS_PORT").unwrap_or_else(|_| "6379".to_string()); - let redis_username = std::env::var("REDIS_USERNAME").unwrap_or_else(|_| "default".to_string()); - let redis_password = std::env::var("REDIS_PASSWORD").unwrap_or_default(); - format!("redis://{redis_username}:{redis_password}@{redis_host}:{redis_port}/{redis_db}") + envvar::redis_uri().unwrap() } #[async_trait] diff --git a/crates/queue/Cargo.toml b/crates/queue/Cargo.toml index 112355eaa..e4938ea24 100644 --- a/crates/queue/Cargo.toml +++ b/crates/queue/Cargo.toml @@ -7,6 +7,7 @@ description = "queue trait with redis and sqs implementations" [dependencies] async-trait = "0.1.73" +envvar = { path = "../envvar" } thiserror = "2.0" tracing = "0.1" # redis diff --git a/crates/queue/src/redis.rs b/crates/queue/src/redis.rs index b5a75666a..e868aa6ed 100644 --- a/crates/queue/src/redis.rs +++ b/crates/queue/src/redis.rs @@ -37,12 +37,7 @@ impl RedisQueue { } fn redis_uri_from_env() -> String { - let redis_db = std::env::var("REDIS_DB").unwrap_or_else(|_| "0".to_string()); - let redis_host = std::env::var("REDIS_HOST").unwrap(); - let redis_port = std::env::var("REDIS_PORT").unwrap_or_else(|_| "6379".to_string()); - let redis_username = std::env::var("REDIS_USERNAME").unwrap_or_else(|_| "default".to_string()); - let redis_password = std::env::var("REDIS_PASSWORD").unwrap_or_default(); - format!("redis://{redis_username}:{redis_password}@{redis_host}:{redis_port}/{redis_db}") + envvar::redis_uri().unwrap() } #[async_trait] diff --git a/crates/uribuilder/Cargo.toml b/crates/uribuilder/Cargo.toml index 7476d7daf..8626e4dca 100644 --- a/crates/uribuilder/Cargo.toml +++ b/crates/uribuilder/Cargo.toml @@ -5,4 +5,5 @@ edition = "2021" rust-version.workspace = true [dependencies] +envvar = { path = "../envvar" } url = "2.5" diff --git a/crates/uribuilder/src/lib.rs b/crates/uribuilder/src/lib.rs index f72a69374..d617fcb67 100644 --- a/crates/uribuilder/src/lib.rs +++ b/crates/uribuilder/src/lib.rs @@ -41,7 +41,7 @@ impl Uri { } pub fn new_from_env_var(env_var: &str) -> Self { - let uri = std::env::var(env_var).unwrap(); + let uri = envvar::required(env_var).unwrap(); Self::new(&uri) } diff --git a/services/auto-confirm/Cargo.toml b/services/auto-confirm/Cargo.toml index 4a864cc6b..a8b802bad 100644 --- a/services/auto-confirm/Cargo.toml +++ b/services/auto-confirm/Cargo.toml @@ -16,6 +16,7 @@ tokio = { version = "1.49", default-features = false, features = [ "macros", ] } unicode-segmentation = "1.12" +envvar = { path = "../../crates/envvar" } types = { path = "../../crates/types" } pg = { path = "../../crates/pg" } service = { path = "../../crates/service" } diff --git a/services/auto-confirm/src/main.rs b/services/auto-confirm/src/main.rs index 2323504c0..f22d4b721 100644 --- a/services/auto-confirm/src/main.rs +++ b/services/auto-confirm/src/main.rs @@ -99,7 +99,7 @@ async fn func(event: LambdaEvent) -> Result) -> Result, rule_instance_id: &str, ) -> Result<(), Box> { - let conn = pool.get_conn().await; + let conn = pool.get_conn().await?; let ri_id: i32 = rule_instance_id.parse()?; let rows = conn.0.query(RULE_INSTANCE_QUERY, &[&ri_id]).await?; diff --git a/services/event/src/events/gdp.rs b/services/event/src/events/gdp.rs index 2d2a3af7a..6eb529621 100644 --- a/services/event/src/events/gdp.rs +++ b/services/event/src/events/gdp.rs @@ -31,7 +31,7 @@ pub async fn handle_gdp( pubsub: &Arc, transaction_id: &str, ) -> Result<(), Box> { - let conn = pool.get_conn().await; + let conn = pool.get_conn().await?; let tx_id: i32 = transaction_id.parse()?; let rows = conn.0.query(GDP_QUERY, &[&tx_id]).await?; diff --git a/services/event/src/events/threshold_profit.rs b/services/event/src/events/threshold_profit.rs index 53d4a8098..19825383d 100644 --- a/services/event/src/events/threshold_profit.rs +++ b/services/event/src/events/threshold_profit.rs @@ -46,7 +46,7 @@ pub async fn handle_threshold_profit( queue: &Arc, transaction_id: &str, ) -> Result<(), Box> { - let conn = pool.get_conn().await; + let conn = pool.get_conn().await?; let tx_id: i32 = transaction_id.parse()?; // get all accounts involved in this transaction diff --git a/services/graphql/Cargo.toml b/services/graphql/Cargo.toml index 075ad61cb..ed9d646eb 100644 --- a/services/graphql/Cargo.toml +++ b/services/graphql/Cargo.toml @@ -19,6 +19,7 @@ aws_lambda_events = { version = "1.0", default-features = false, features = [ ] } tower-http = { version = "0.6", features = ["cors"] } http = "1.4" +envvar = { path = "../../crates/envvar" } shutdown = { path = "../../crates/shutdown" } futures-util = "0.3" wsclient = { path = "../../crates/wsclient" } diff --git a/services/measure/Cargo.toml b/services/measure/Cargo.toml index 4d0b3653e..0074ec9dd 100644 --- a/services/measure/Cargo.toml +++ b/services/measure/Cargo.toml @@ -9,6 +9,7 @@ axum = { version = "0.8", features = ["ws"] } tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } tokio = { version = "1.49", features = ["macros", "rt-multi-thread"] } +envvar = { path = "../../crates/envvar" } shutdown = { path = "../../crates/shutdown" } futures = "0.3" serde = { version = "1.0", features = ["serde_derive"] } diff --git a/services/request-approve/Cargo.toml b/services/request-approve/Cargo.toml index b98fcdb0f..509377a80 100644 --- a/services/request-approve/Cargo.toml +++ b/services/request-approve/Cargo.toml @@ -9,6 +9,7 @@ axum = "0.8" tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } tokio = { version = "1.49", features = ["macros", "rt-multi-thread"] } +envvar = { path = "../../crates/envvar" } shutdown = { path = "../../crates/shutdown" } pg = { path = "../../crates/pg" } service = { path = "../../crates/service" } diff --git a/services/request-by-id/Cargo.toml b/services/request-by-id/Cargo.toml index 528d6da24..7b1ca9475 100644 --- a/services/request-by-id/Cargo.toml +++ b/services/request-by-id/Cargo.toml @@ -9,6 +9,7 @@ axum = "0.8" tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } tokio = { version = "1.49", features = ["macros", "rt-multi-thread"] } +envvar = { path = "../../crates/envvar" } shutdown = { path = "../../crates/shutdown" } pg = { path = "../../crates/pg" } service = { path = "../../crates/service" } diff --git a/services/request-create/Cargo.toml b/services/request-create/Cargo.toml index 7895b2cd4..4da4699a2 100644 --- a/services/request-create/Cargo.toml +++ b/services/request-create/Cargo.toml @@ -13,6 +13,7 @@ tokio = { version = "1.49", features = ["macros", "rt-multi-thread"] } thiserror = "2.0" shutdown = { path = "../../crates/shutdown" } pg = { path = "../../crates/pg" } +envvar = { path = "../../crates/envvar" } service = { path = "../../crates/service" } types = { path = "../../crates/types" } httpclient = { path = "../../crates/httpclient" } diff --git a/services/requests-by-account/Cargo.toml b/services/requests-by-account/Cargo.toml index 5bd6df132..de16b2ae0 100644 --- a/services/requests-by-account/Cargo.toml +++ b/services/requests-by-account/Cargo.toml @@ -9,6 +9,7 @@ axum = "0.8" tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } tokio = { version = "1.49", features = ["macros", "rt-multi-thread"] } +envvar = { path = "../../crates/envvar" } shutdown = { path = "../../crates/shutdown" } pg = { path = "../../crates/pg" } service = { path = "../../crates/service" } diff --git a/services/rule/Cargo.toml b/services/rule/Cargo.toml index 8195b911f..088f76739 100644 --- a/services/rule/Cargo.toml +++ b/services/rule/Cargo.toml @@ -18,6 +18,7 @@ types = { path = "../../crates/types" } pg = { path = "../../crates/pg" } chrono = "0.4" nanoid = "0.4.0" +envvar = { path = "../../crates/envvar" } service = { path = "../../crates/service" } shutdown = { path = "../../crates/shutdown" } cache = { path = "../../crates/cache" } diff --git a/services/transaction-by-id/Cargo.toml b/services/transaction-by-id/Cargo.toml index 113f62963..9015193f8 100644 --- a/services/transaction-by-id/Cargo.toml +++ b/services/transaction-by-id/Cargo.toml @@ -9,6 +9,7 @@ axum = "0.8" tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } tokio = { version = "1.49", features = ["macros", "rt-multi-thread"] } +envvar = { path = "../../crates/envvar" } shutdown = { path = "../../crates/shutdown" } pg = { path = "../../crates/pg" } service = { path = "../../crates/service" } diff --git a/services/transactions-by-account/Cargo.toml b/services/transactions-by-account/Cargo.toml index fd27f9972..60d9a432c 100644 --- a/services/transactions-by-account/Cargo.toml +++ b/services/transactions-by-account/Cargo.toml @@ -9,6 +9,7 @@ axum = "0.8" tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } tokio = { version = "1.49", features = ["macros", "rt-multi-thread"] } +envvar = { path = "../../crates/envvar" } shutdown = { path = "../../crates/shutdown" } pg = { path = "../../crates/pg" } service = { path = "../../crates/service" } From 5f56a58c6bccf9a04b8f6350c57c1c3159cb1739 Mon Sep 17 00:00:00 2001 From: max funk Date: Tue, 3 Mar 2026 07:49:24 -0800 Subject: [PATCH 04/12] envvar crate lockfile --- Cargo.lock | 28 ++++++++++++++++++++++++++++ 1 file changed, 28 insertions(+) diff --git a/Cargo.lock b/Cargo.lock index 49dfef670..3c6f21689 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -365,6 +365,7 @@ name = "auto-confirm" version = "0.1.0" dependencies = [ "aws_lambda_events", + "envvar", "fakeit", "lambda_runtime", "openssl", @@ -384,6 +385,7 @@ name = "auto-transact" version = "0.1.0" dependencies = [ "axum", + "envvar", "httpclient", "openssl", "pg", @@ -883,6 +885,7 @@ name = "balance-by-account" version = "0.1.0" dependencies = [ "axum", + "envvar", "openssl", "pg", "service", @@ -1077,6 +1080,7 @@ dependencies = [ "async-trait", "aws-config", "aws-sdk-dynamodb", + "envvar", "fred", "serde_json", "thiserror 2.0.17", @@ -1464,6 +1468,10 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "envvar" +version = "0.1.0" + [[package]] name = "equivalent" version = "1.0.1" @@ -1858,6 +1866,7 @@ dependencies = [ "async-stream", "aws_lambda_events", "axum", + "envvar", "futures-util", "http 1.4.0", "httpclient", @@ -2083,6 +2092,8 @@ dependencies = [ "httpmock", "reqwest 0.13.1", "serde", + "serde_json", + "thiserror 2.0.17", "tokio", ] @@ -2648,6 +2659,7 @@ name = "measure" version = "0.1.0" dependencies = [ "axum", + "envvar", "futures", "http 1.4.0", "pg", @@ -3026,6 +3038,7 @@ dependencies = [ "bb8-postgres", "chrono", "dotenvy", + "envvar", "geo-types", "mockall", "regex", @@ -3036,6 +3049,7 @@ dependencies = [ "serial_test", "tokio", "tokio-postgres", + "tracing", "types", ] @@ -3275,6 +3289,7 @@ name = "pubsub" version = "0.1.0" dependencies = [ "async-trait", + "envvar", "fred", "thiserror 2.0.17", "tokio", @@ -3299,6 +3314,7 @@ dependencies = [ "async-trait", "aws-config", "aws-sdk-sqs", + "envvar", "fred", "thiserror 2.0.17", "tracing", @@ -3527,6 +3543,7 @@ name = "request-approve" version = "0.1.0" dependencies = [ "axum", + "envvar", "openssl", "pg", "service", @@ -3542,6 +3559,7 @@ name = "request-by-id" version = "0.1.0" dependencies = [ "axum", + "envvar", "openssl", "pg", "service", @@ -3558,6 +3576,7 @@ version = "0.1.0" dependencies = [ "axum", "cache", + "envvar", "httpclient", "openssl", "pg", @@ -3577,6 +3596,7 @@ name = "requests-by-account" version = "0.1.0" dependencies = [ "axum", + "envvar", "openssl", "pg", "service", @@ -3716,6 +3736,7 @@ dependencies = [ "axum", "cache", "chrono", + "envvar", "nanoid", "openssl", "pg", @@ -4129,11 +4150,15 @@ dependencies = [ name = "service" version = "0.1.0" dependencies = [ + "axum", "cache", "cognitoidp", + "envvar", + "http-body-util", "mockall", "pg", "rust_decimal", + "serde_json", "tokio", "types", ] @@ -4856,6 +4881,7 @@ name = "transaction-by-id" version = "0.1.0" dependencies = [ "axum", + "envvar", "openssl", "pg", "service", @@ -4871,6 +4897,7 @@ name = "transactions-by-account" version = "0.1.0" dependencies = [ "axum", + "envvar", "openssl", "pg", "service", @@ -4992,6 +5019,7 @@ checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1" name = "uribuilder" version = "0.1.0" dependencies = [ + "envvar", "url", ] From 3bd151417d56880876386b3109f2550ba49d546e Mon Sep 17 00:00:00 2001 From: max funk Date: Tue, 3 Mar 2026 07:51:43 -0800 Subject: [PATCH 05/12] client error enum --- crates/httpclient/Cargo.toml | 2 + crates/httpclient/src/lib.rs | 98 ++++++++++++++++++++++++++++-------- 2 files changed, 78 insertions(+), 22 deletions(-) diff --git a/crates/httpclient/Cargo.toml b/crates/httpclient/Cargo.toml index 3a33799bd..1f55a6534 100644 --- a/crates/httpclient/Cargo.toml +++ b/crates/httpclient/Cargo.toml @@ -11,6 +11,8 @@ aws-sigv4 = "1.3" aws-smithy-runtime-api = "1.10" http = "1.4" reqwest = { version = "0.13", features = ["json"] } +serde_json = "1" +thiserror = "2.0" [dev-dependencies] httpmock = "0.8" diff --git a/crates/httpclient/src/lib.rs b/crates/httpclient/src/lib.rs index bae06a505..ab64fe539 100644 --- a/crates/httpclient/src/lib.rs +++ b/crates/httpclient/src/lib.rs @@ -2,9 +2,29 @@ use aws_credential_types::provider::ProvideCredentials; use aws_sigv4::http_request::{sign, SignableBody, SignableRequest, SigningSettings}; use aws_sigv4::sign::v4; use aws_smithy_runtime_api::client::identity::Identity; -use reqwest::{Client, Error, Response}; +use reqwest::Client; use std::{env, time::SystemTime}; +#[derive(Debug, thiserror::Error)] +pub enum ClientError { + #[error("request error: {0}")] + Request(#[from] reqwest::Error), + #[error("signing error: {0}")] + Signing(String), + #[error("aws config error: {0}")] + AwsConfig(String), + #[error("request build error: {0}")] + Build(String), + #[error("{message}")] + Downstream { status: u16, message: String }, +} + +impl From for ClientError { + fn from(s: String) -> Self { + ClientError::Build(s) + } +} + #[derive(Default)] pub struct HttpClient(Client); @@ -13,32 +33,51 @@ impl HttpClient { Self::default() } - pub async fn post(&self, url: String, body: String) -> Result { + pub async fn post(&self, url: String, body: String) -> Result { let mut http_request = http::Request::builder() .method("POST") .uri(url) - .header( - "Content-Type", - http::HeaderValue::from_str("application/json").unwrap(), - ) + .header("Content-Type", "application/json") .body(body) - .unwrap(); + .map_err(|e| ClientError::Build(e.to_string()))?; // sign request when testing lambda if env::var("AWS_LAMBDA_FUNCTION_NAME").ok().is_some() { - HttpClient::sign(&mut http_request).await; + HttpClient::sign(&mut http_request).await?; } - let req = reqwest::Request::try_from(http_request).unwrap(); - - self.0.execute(req).await + let req = reqwest::Request::try_from(http_request) + .map_err(|e| ClientError::Build(e.to_string()))?; + + let response = self.0.execute(req).await?; + let status = response.status(); + let text = response.text().await?; + if !status.is_success() { + let msg = serde_json::from_str::(&text) + .ok() + .and_then(|v| v.get("error").and_then(|e| e.as_str()).map(String::from)) + .unwrap_or(text); + return Err(ClientError::Downstream { + status: status.as_u16(), + message: msg, + }); + } + Ok(text) } - async fn sign(http_request: &mut http::Request) { + async fn sign(http_request: &mut http::Request) -> Result<(), ClientError> { let config = aws_config::load_defaults(aws_config::BehaviorVersion::latest()).await; - let region = config.region().unwrap().as_ref(); - let provider = config.credentials_provider().unwrap(); - let credentials = provider.provide_credentials().await.unwrap(); + let region = config + .region() + .ok_or_else(|| ClientError::AwsConfig("missing aws region".into()))? + .as_ref(); + let provider = config + .credentials_provider() + .ok_or_else(|| ClientError::AwsConfig("missing credentials provider".into()))?; + let credentials = provider + .provide_credentials() + .await + .map_err(|e| ClientError::AwsConfig(e.to_string()))?; // https://github.com/awslabs/aws-sdk-rust/discussions/868 let identity = Identity::new(credentials, None); let signing_settings = SigningSettings::default(); @@ -50,7 +89,7 @@ impl HttpClient { .time(SystemTime::now()) .settings(signing_settings) .build() - .unwrap() + .map_err(|e| ClientError::Signing(e.to_string()))? .into(); let signable_request = SignableRequest::new( @@ -59,15 +98,17 @@ impl HttpClient { http_request .headers() .iter() - .map(|(k, v)| (k.as_str(), v.to_str().unwrap())), + .map(|(k, v)| (k.as_str(), v.to_str().unwrap_or_default())), SignableBody::Bytes(http_request.body().as_bytes()), ) - .unwrap(); + .map_err(|e| ClientError::Signing(e.to_string()))?; let (signing_instructions, _signature) = sign(signable_request, &signing_params) - .unwrap() + .map_err(|e| ClientError::Signing(e.to_string()))? .into_parts(); signing_instructions.apply_to_request_http1x(http_request); + + Ok(()) } } @@ -83,6 +124,19 @@ mod tests { message: String, } + #[tokio::test] + async fn it_returns_client_error_on_connection_refused() { + let client = HttpClient::new(); + let result = client + .post("http://localhost:1".to_string(), String::new()) + .await; + assert!(result.is_err()); + assert!(matches!( + result.unwrap_err(), + crate::ClientError::Request(_) + )); + } + #[tokio::test] async fn it_receives_a_response() { let server = MockServer::start(); @@ -100,10 +154,10 @@ mod tests { let url = server.base_url() + "/"; let test_client = HttpClient::new(); - let response = test_client.post(url, String::from("")).await.unwrap(); - let response_body: TestMessage = response.json().await.unwrap(); + let response_body = test_client.post(url, String::from("")).await.unwrap(); + let parsed: TestMessage = serde_json::from_str(&response_body).unwrap(); - let got = response_body.message; + let got = parsed.message; let want = "test".to_string(); assert_eq!(got, want); From 58cbcde1b681d4d143bcabb9a289b662bb7b1ac7 Mon Sep 17 00:00:00 2001 From: max funk Date: Tue, 3 Mar 2026 07:52:47 -0800 Subject: [PATCH 06/12] service error struct --- crates/service/Cargo.toml | 4 ++ crates/service/src/lib.rs | 89 +++++++++++++++++++++++++++++++++++++-- 2 files changed, 90 insertions(+), 3 deletions(-) diff --git a/crates/service/Cargo.toml b/crates/service/Cargo.toml index 3192b88b2..ee32b1a9f 100644 --- a/crates/service/Cargo.toml +++ b/crates/service/Cargo.toml @@ -8,10 +8,14 @@ description = "service" [dependencies] types = { path = "../types" } rust_decimal = { version = "1.40", default-features = false } +envvar = { path = "../envvar" } pg = { path = "../pg" } cognitoidp = { path = "../cognitoidp" } cache = { path = "../cache" } +axum = "0.8" +serde_json = "1" [dev-dependencies] mockall = "0.14" tokio = { version = "1.49", default-features = false, features = ["macros"] } +http-body-util = "0.1" diff --git a/crates/service/src/lib.rs b/crates/service/src/lib.rs index 0c13913f5..db61e1c17 100644 --- a/crates/service/src/lib.rs +++ b/crates/service/src/lib.rs @@ -1,3 +1,5 @@ +use axum::http::StatusCode; +use axum::response::{IntoResponse, Response}; use cache::Cache; use cognitoidp::CognitoJwkSet; use pg::model::ModelTrait; @@ -15,6 +17,47 @@ use types::{ transaction_item::TransactionItems, }; +pub struct ServiceError { + status: StatusCode, + message: String, +} + +impl ServiceError { + pub fn bad_request(message: &str) -> Self { + Self { + status: StatusCode::BAD_REQUEST, + message: message.to_string(), + } + } + + pub fn not_found(message: &str) -> Self { + Self { + status: StatusCode::NOT_FOUND, + message: message.to_string(), + } + } + + pub fn internal(message: &str) -> Self { + Self { + status: StatusCode::INTERNAL_SERVER_ERROR, + message: message.to_string(), + } + } +} + +impl From for ServiceError { + fn from(s: String) -> Self { + ServiceError::internal(&s) + } +} + +impl IntoResponse for ServiceError { + fn into_response(self) -> Response { + let body = serde_json::json!({ "error": self.message }); + (self.status, axum::Json(body)).into_response() + } +} + pub struct Service<'a, T: ModelTrait> { conn: &'a T, cache: Option>, @@ -177,7 +220,7 @@ impl<'a, T: ModelTrait> Service<'a, T> { let debitor_account_balance = debitor_funds_available .get_account_balance(&debitor) - .unwrap(); + .ok_or_else(|| format!("account balance not found for {debitor}"))?; if !debitor_account_balance.sufficient_balance(funds_required) { let err_msg = &*format!( @@ -279,7 +322,12 @@ impl<'a, T: ModelTrait> Service<'a, T> { }, } - let transaction_id = request.clone().id.unwrap().parse::().unwrap(); + let transaction_id = request + .clone() + .id + .ok_or("missing transaction id")? + .parse::() + .map_err(|e| format!("invalid transaction id: {e}"))?; // add approval time to transaction if not previously approved if approval_time.is_none() { @@ -464,7 +512,7 @@ impl<'a, T: ModelTrait> Service<'a, T> { pub async fn get_json_web_key_set(&self) -> Result, Box> { if env::var("ENABLE_API_AUTH") == Ok("true".to_string()) { - let uri = env::var("COGNITO_JWKS_URI").expect("msg: COGNITO_JWKS_URI not set"); + let uri = envvar::required("COGNITO_JWKS_URI").unwrap(); let jwks = CognitoJwkSet::new(uri.as_str()).await; match jwks { Ok(jwks) => Ok(Some(jwks)), @@ -663,6 +711,8 @@ impl<'a, T: ModelTrait> Service<'a, T> { mod tests { use std::{str::FromStr, vec}; + use axum::response::IntoResponse; + use http_body_util::BodyExt; use mockall::predicate; use pg::model::MockModelTrait; use rust_decimal::Decimal; @@ -680,6 +730,39 @@ mod tests { transaction_item::{TransactionItem, TransactionItems}, }; + #[tokio::test] + async fn it_returns_400_with_json_error_for_bad_request() { + let err = super::ServiceError::bad_request("test error"); + let response = err.into_response(); + assert_eq!(response.status(), axum::http::StatusCode::BAD_REQUEST); + let body = response.into_body().collect().await.unwrap().to_bytes(); + let json: serde_json::Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(json["error"], "test error"); + } + + #[tokio::test] + async fn it_returns_404_with_json_error_for_not_found() { + let err = super::ServiceError::not_found("missing resource"); + let response = err.into_response(); + assert_eq!(response.status(), axum::http::StatusCode::NOT_FOUND); + let body = response.into_body().collect().await.unwrap().to_bytes(); + let json: serde_json::Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(json["error"], "missing resource"); + } + + #[tokio::test] + async fn it_returns_500_with_json_error_for_internal() { + let err = super::ServiceError::internal("server failure"); + let response = err.into_response(); + assert_eq!( + response.status(), + axum::http::StatusCode::INTERNAL_SERVER_ERROR + ); + let body = response.into_body().collect().await.unwrap().to_bytes(); + let json: serde_json::Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(json["error"], "server failure"); + } + #[tokio::test] async fn create_account_queries_by_account() { let mut conn = MockModelTrait::new(); From 67425140729264450bba9baaf0ef8ffde61fd567 Mon Sep 17 00:00:00 2001 From: max funk Date: Tue, 3 Mar 2026 07:55:52 -0800 Subject: [PATCH 07/12] log and return result --- crates/pg/Cargo.toml | 2 ++ crates/pg/src/postgres.rs | 26 +++++++++++++------------- 2 files changed, 15 insertions(+), 13 deletions(-) diff --git a/crates/pg/Cargo.toml b/crates/pg/Cargo.toml index 92f207eb6..5674b6901 100644 --- a/crates/pg/Cargo.toml +++ b/crates/pg/Cargo.toml @@ -12,11 +12,13 @@ tokio-postgres = { version = "0.7", features = [ "with-chrono-0_4", "with-geo-types-0_7", ] } +envvar = { path = "../envvar" } types = { path = "../types" } serde = { version = "1.0", features = ["derive"] } serde-aux = { version = "4.7", default-features = false } rust_decimal = { version = "1.40", features = ["db-tokio-postgres"] } geo-types = "0.7" +tracing = "0.1" chrono = "0.4" mockall = "0.14" diff --git a/crates/pg/src/postgres.rs b/crates/pg/src/postgres.rs index cbd5e39b2..a1ac5bd99 100644 --- a/crates/pg/src/postgres.rs +++ b/crates/pg/src/postgres.rs @@ -1,17 +1,16 @@ use bb8::{Pool, PooledConnection}; use bb8_postgres::PostgresConnectionManager; -use std::env; use tokio_postgres::{types::ToSql, NoTls, Row}; pub struct DB; impl DB { pub fn create_conn_uri_from_env_vars() -> String { - let pguser = DB::get_env_var("PGUSER"); - let pgpassword = DB::get_env_var("PGPASSWORD"); - let pghost = DB::get_env_var("PGHOST"); - let pgport = DB::get_env_var("PGPORT"); - let pgdatabase = DB::get_env_var("PGDATABASE"); + let pguser = envvar::required("PGUSER").unwrap(); + let pgpassword = envvar::required("PGPASSWORD").unwrap(); + let pghost = envvar::required("PGHOST").unwrap(); + let pgport = envvar::required("PGPORT").unwrap(); + let pgdatabase = envvar::required("PGDATABASE").unwrap(); format!("postgresql://{pguser}:{pgpassword}@{pghost}:{pgport}/{pgdatabase}") } @@ -20,10 +19,6 @@ impl DB { let pool = Pool::builder().build(manager).await.unwrap(); ConnectionPool(pool) } - - pub fn get_env_var(var_name: &str) -> String { - env::var(var_name).unwrap_or_else(|_| panic!("{var_name} not set")) - } } // https://github.com/tokio-rs/axum/blob/5793e75aacfeae16f02fea144ecc2ee7dcb12f55/examples/tokio-postgres/src/main.rs @@ -31,9 +26,14 @@ impl DB { pub struct ConnectionPool(pub Pool>); impl ConnectionPool { - pub async fn get_conn(&self) -> DatabaseConnection { - let conn = self.0.get_owned().await.unwrap(); // todo: handle error - DatabaseConnection(conn) + pub async fn get_conn( + &self, + ) -> Result> { + let conn = self.0.get_owned().await.map_err(|e| { + tracing::error!("failed to get db connection: {}", e); + e + })?; + Ok(DatabaseConnection(conn)) } } From 74d1dd21c78df0c5b8ea4cdd16f0f8102aa45b39 Mon Sep 17 00:00:00 2001 From: max funk Date: Tue, 3 Mar 2026 08:00:44 -0800 Subject: [PATCH 08/12] return errors --- crates/pg/src/model.rs | 116 +++++----- services/auto-transact/src/main.rs | 26 +-- services/balance-by-account/src/main.rs | 14 +- services/event/src/main.rs | 6 +- services/graphql/src/main.rs | 210 +++++++++++-------- services/measure/src/main.rs | 9 +- services/request-approve/src/main.rs | 33 +-- services/request-by-id/src/main.rs | 14 +- services/request-create/src/main.rs | 79 ++++--- services/requests-by-account/src/main.rs | 18 +- services/rule/src/lib.rs | 16 +- services/rule/src/main.rs | 77 ++++--- services/rule/src/rules/approval.rs | 119 ++++++++++- services/rule/src/rules/transaction_item.rs | 75 ++++++- services/transaction-by-id/src/main.rs | 14 +- services/transactions-by-account/src/main.rs | 18 +- 16 files changed, 542 insertions(+), 302 deletions(-) diff --git a/crates/pg/src/model.rs b/crates/pg/src/model.rs index e4452d5a2..0820df2ad 100644 --- a/crates/pg/src/model.rs +++ b/crates/pg/src/model.rs @@ -221,14 +221,19 @@ impl ModelTrait for DatabaseConnection { // add creditor revenue string as second param values.push_param(tr_item.clone().revenue_string()); // add transaction item id as third param - values.push_param(tr_item.id.clone().unwrap().parse::().unwrap()); + let tr_item_id: i32 = tr_item + .id + .clone() + .ok_or("missing id in transaction item")? + .parse()?; + values.push_param(tr_item_id); // add debitor account as fourth param values.push_param(tr_item.clone().debitor); // add debitor expense string as fifth param values.push_param(tr_item.clone().expense_string()); // add transaction item id as sixth param - values.push_param(tr_item.clone().id.unwrap().parse::().unwrap()); + values.push_param(tr_item_id); } let table = crate::sqls::balance::AccountBalanceTable::new(); @@ -465,14 +470,12 @@ impl ModelTrait for DatabaseConnection { let sql = table.fn_select_insert_transaction_sql(approval_lengths); // convert rust transaction values to postgres values - let transaction_rule_instance_id = parse_pg_int4(transaction.rule_instance_id).unwrap(); - let transaction_author_device_latlng = - parse_pg_point(transaction.author_device_latlng).unwrap(); - let transaction_equilibrium_time = - parse_pg_timestamp(transaction.equilibrium_time).unwrap(); - let transaction_sum_value = parse_pg_numeric(Some(transaction.sum_value)).unwrap(); - let transaction_created_at = parse_pg_timestamp(None::).unwrap(); - let transaction_event_time = parse_pg_timestamp(None::).unwrap(); + let transaction_rule_instance_id = parse_pg_int4(transaction.rule_instance_id)?; + let transaction_author_device_latlng = parse_pg_point(transaction.author_device_latlng)?; + let transaction_equilibrium_time = parse_pg_timestamp(transaction.equilibrium_time)?; + let transaction_sum_value = parse_pg_numeric(Some(transaction.sum_value))?; + let transaction_created_at = parse_pg_timestamp(None::)?; + let transaction_event_time = parse_pg_timestamp(None::)?; let mut values = to_sql_vec![ None::, // id @@ -497,59 +500,61 @@ impl ModelTrait for DatabaseConnection { values.push_param(tr_item.item_id); // item_id - let tr_item_price = parse_pg_numeric(Some(tr_item.price)).unwrap(); + let tr_item_price = parse_pg_numeric(Some(tr_item.price))?; values.push_param(tr_item_price); // price - let tr_item_quantity = parse_pg_numeric(Some(tr_item.quantity)).unwrap(); + let tr_item_quantity = parse_pg_numeric(Some(tr_item.quantity))?; values.push_param(tr_item_quantity); // quantity - let tr_item_rule_instance_id = parse_pg_int4(tr_item.rule_instance_id).unwrap(); + let tr_item_rule_instance_id = parse_pg_int4(tr_item.rule_instance_id)?; values.push_param(tr_item_rule_instance_id); // rule_instance_id values.push_param(tr_item.rule_exec_ids); // rule_exec_ids values.push_param(tr_item.unit_of_measurement); // unit_of_measurement - let tr_item_units_measured = parse_pg_numeric(tr_item.units_measured).unwrap(); + let tr_item_units_measured = parse_pg_numeric(tr_item.units_measured)?; values.push_param(tr_item_units_measured); // units_measured values.push_param(tr_item.debitor); // debitor values.push_param(tr_item.creditor); // creditor - let debitor_profile_id = parse_pg_int4(tr_item.debitor_profile_id).unwrap(); + let debitor_profile_id = parse_pg_int4(tr_item.debitor_profile_id)?; values.push_param(debitor_profile_id); // debitor_profile_id - let creditor_profile_id = parse_pg_int4(tr_item.creditor_profile_id).unwrap(); + let creditor_profile_id = parse_pg_int4(tr_item.creditor_profile_id)?; values.push_param(creditor_profile_id); // creditor_profile_id - let tr_item_debitor_approval_time = - parse_pg_timestamp(tr_item.debitor_approval_time).unwrap(); + let tr_item_debitor_approval_time = parse_pg_timestamp(tr_item.debitor_approval_time)?; values.push_param(tr_item_debitor_approval_time); // debitor_approval_time let tr_item_creditor_approval_time = - parse_pg_timestamp(tr_item.creditor_approval_time).unwrap(); + parse_pg_timestamp(tr_item.creditor_approval_time)?; values.push_param(tr_item_creditor_approval_time); // creditor_approval_time let tr_item_debitor_expiration_time = - parse_pg_timestamp(tr_item.debitor_expiration_time).unwrap(); + parse_pg_timestamp(tr_item.debitor_expiration_time)?; values.push_param(tr_item_debitor_expiration_time); // debitor_expiration_time let tr_item_creditor_expiration_time = - parse_pg_timestamp(tr_item.creditor_expiration_time).unwrap(); + parse_pg_timestamp(tr_item.creditor_expiration_time)?; values.push_param(tr_item_creditor_expiration_time); // creditor_expiration_time let tr_item_creditor_rejection_time = - parse_pg_timestamp(tr_item.creditor_rejection_time).unwrap(); + parse_pg_timestamp(tr_item.creditor_rejection_time)?; values.push_param(tr_item_creditor_rejection_time); // creditor_rejection_time let tr_item_debitor_rejection_time = - parse_pg_timestamp(tr_item.debitor_rejection_time).unwrap(); + parse_pg_timestamp(tr_item.debitor_rejection_time)?; values.push_param(tr_item_debitor_rejection_time); // debitor_rejection_time - for approval in tr_item.approvals.unwrap().into_iter() { + let approvals = tr_item + .approvals + .ok_or("missing approvals in transaction item")?; + for approval in approvals.into_iter() { values.push_param(None::); // id - let approval_rule_instance_id = parse_pg_int4(approval.rule_instance_id).unwrap(); + let approval_rule_instance_id = parse_pg_int4(approval.rule_instance_id)?; values.push_param(approval_rule_instance_id); // rule_instance_id values.push_param(None::); // transaction_id @@ -558,17 +563,16 @@ impl ModelTrait for DatabaseConnection { values.push_param(approval.account_role); // account_role values.push_param(approval.device_id); // device_id - let approval_device_latlng = parse_pg_point(approval.device_latlng).unwrap(); + let approval_device_latlng = parse_pg_point(approval.device_latlng)?; values.push_param(approval_device_latlng); // device_latlng - let approval_approval_time = parse_pg_timestamp(approval.approval_time).unwrap(); + let approval_approval_time = parse_pg_timestamp(approval.approval_time)?; values.push_param(approval_approval_time); // approval_time - let approval_rejection_time = parse_pg_timestamp(approval.rejection_time).unwrap(); + let approval_rejection_time = parse_pg_timestamp(approval.rejection_time)?; values.push_param(approval_rejection_time); // rejection_time - let approval_expiration_time = - parse_pg_timestamp(approval.expiration_time).unwrap(); + let approval_expiration_time = parse_pg_timestamp(approval.expiration_time)?; values.push_param(approval_expiration_time); // expiration_time } } @@ -849,43 +853,45 @@ impl ModelTrait for DatabaseConnection { } fn parse_pg_int4(s: Option) -> Result, Box> { - if s.clone().is_none() || s.clone().unwrap() == *"" { - return Ok(None); + match s { + None => Ok(None), + Some(ref v) if v.is_empty() => Ok(None), + Some(v) => Ok(Some(v.parse::()?)), } - // test for non-numeric string - if s.clone().unwrap().parse::().is_err() { - return Err("non-numeric string".into()); - } - Ok(Some(s.unwrap().parse::().unwrap())) } fn parse_pg_point(s: Option) -> Result>, Box> { - if s.clone().is_none() || s.clone().unwrap() == *"" { - return Ok(None); - } - // create point from "(39.534552,-119.737825)" - let stripped_parens = s.unwrap().replace(['(', ')'], ""); - let coords: Vec<&str> = stripped_parens.split(',').collect(); - let lat = coords[0].trim().parse::().unwrap(); - let lng = coords[1].trim().parse::().unwrap(); - let point = Point::new(lat, lng); - Ok(Some(point)) + match s { + None => Ok(None), + Some(ref v) if v.is_empty() => Ok(None), + Some(v) => { + // create point from "(39.534552,-119.737825)" + let stripped_parens = v.replace(['(', ')'], ""); + let coords: Vec<&str> = stripped_parens.split(',').collect(); + let lat = coords[0].trim().parse::()?; + let lng = coords[1].trim().parse::()?; + Ok(Some(Point::new(lat, lng))) + } + } } fn parse_pg_timestamp(time: Option) -> Result>, Box> { - if time.clone().is_none() { - return Ok(None::>); + match time { + None => Ok(None), + Some(t) => Ok(Some(t.0)), } - Ok(Some(time.unwrap().0)) } fn parse_pg_numeric(s: Option) -> Result, Box> { - if s.clone().is_none() || s.clone().unwrap() == *"" { - return Ok(None); + match s { + None => Ok(None), + Some(ref v) if v.is_empty() => Ok(None), + Some(v) => { + let mut decimal = Decimal::from_str_exact(v.as_str())?; + decimal.rescale(FIXED_DECIMAL_PLACES as u32); + Ok(Some(decimal)) + } } - let mut decimal = Decimal::from_str_exact(s.unwrap().as_str())?; - decimal.rescale(FIXED_DECIMAL_PLACES as u32); - Ok(Some(decimal)) } #[cfg(test)] diff --git a/services/auto-transact/src/main.rs b/services/auto-transact/src/main.rs index 3a7acf2e7..d924e25c9 100644 --- a/services/auto-transact/src/main.rs +++ b/services/auto-transact/src/main.rs @@ -16,12 +16,10 @@ use uribuilder::Uri; const READINESS_CHECK_PATH: &str = "READINESS_CHECK_PATH"; fn retry_config(service_name: &str) -> RetryConfig { - let max_retries: u32 = env::var("AUTO_TRANSACT_MAX_RETRIES") - .unwrap_or("3".to_string()) + let max_retries: u32 = envvar::optional("AUTO_TRANSACT_MAX_RETRIES", "3") .parse() .unwrap_or(3); - let delay_ms: u64 = env::var("AUTO_TRANSACT_RETRY_DELAY_MS") - .unwrap_or("1000".to_string()) + let delay_ms: u64 = envvar::optional("AUTO_TRANSACT_RETRY_DELAY_MS", "1000") .parse() .unwrap_or(1000); RetryConfig { @@ -39,16 +37,11 @@ async fn process_transaction(transaction: Transaction) -> Result<(), String> { let rule_body = serde_json::to_string(&transaction) .map_err(|e| format!("serialize for rule failed: {}", e))?; - let rule_response = client + let rule_text = client .post(rule_uri, rule_body) .await .map_err(|e| format!("rule request failed: {:?}", e))?; - let rule_text = rule_response - .text() - .await - .map_err(|e| format!("rule response read failed: {}", e))?; - let rule_applied: IntraTransaction = serde_json::from_str(&rule_text) .map_err(|e| format!("rule response parse failed: {}", e))?; @@ -69,13 +62,9 @@ async fn process_transaction(transaction: Transaction) -> Result<(), String> { let uri = create_uri.clone(); let body = create_body.clone(); async move { - let response = client + client .post(uri, body) .await - .map_err(|e| -> Box { Box::new(e) })?; - response - .text() - .await .map_err(|e| -> Box { Box::new(e) }) } }) @@ -180,8 +169,7 @@ async fn queue_worker(queue: Arc) { async fn main() { tracing_subscriber::fmt::init(); - let readiness_check_path = env::var(READINESS_CHECK_PATH) - .unwrap_or_else(|_| panic!("{READINESS_CHECK_PATH} variable assignment")); + let readiness_check_path = envvar::required(READINESS_CHECK_PATH).unwrap(); let conn_uri = DB::create_conn_uri_from_env_vars(); let pool = DB::new_pool(&conn_uri).await; @@ -205,8 +193,8 @@ async fn main() { ) .with_state(pool); - let hostname_or_ip = env::var("HOSTNAME_OR_IP").unwrap_or("0.0.0.0".to_string()); - let port = env::var("AUTO_TRANSACT_PORT").unwrap(); + let hostname_or_ip = envvar::optional("HOSTNAME_OR_IP", "0.0.0.0"); + let port = envvar::required("AUTO_TRANSACT_PORT").unwrap(); let serve_addr = format!("{hostname_or_ip}:{port}"); let mut addrs_iter = serve_addr.to_socket_addrs().unwrap_or( diff --git a/services/balance-by-account/src/main.rs b/services/balance-by-account/src/main.rs index 26fca668d..220cfacf0 100644 --- a/services/balance-by-account/src/main.rs +++ b/services/balance-by-account/src/main.rs @@ -7,7 +7,7 @@ use axum::{ use pg::postgres::{ConnectionPool, DB}; use service::Service; use shutdown::shutdown_signal; -use std::{env, net::ToSocketAddrs}; +use std::net::ToSocketAddrs; use tokio::net::TcpListener; use types::request_response::QueryByAccount; @@ -20,7 +20,10 @@ async fn handle_event( ) -> Result { let client_request = event.0; - let conn = pool.get_conn().await; + let conn = pool + .get_conn() + .await + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; let svc = Service::new(&conn, None); @@ -38,8 +41,7 @@ async fn handle_event( async fn main() { tracing_subscriber::fmt::init(); - let readiness_check_path = env::var(READINESS_CHECK_PATH) - .unwrap_or_else(|_| panic!("{READINESS_CHECK_PATH} variable assignment")); + let readiness_check_path = envvar::required(READINESS_CHECK_PATH).unwrap(); let conn_uri = DB::create_conn_uri_from_env_vars(); @@ -53,9 +55,9 @@ async fn main() { ) .with_state(pool); - let hostname_or_ip = env::var("HOSTNAME_OR_IP").unwrap_or("0.0.0.0".to_string()); + let hostname_or_ip = envvar::optional("HOSTNAME_OR_IP", "0.0.0.0"); - let port = env::var("BALANCE_BY_ACCOUNT_PORT").unwrap(); + let port = envvar::required("BALANCE_BY_ACCOUNT_PORT").unwrap(); let serve_addr = format!("{hostname_or_ip}:{port}"); diff --git a/services/event/src/main.rs b/services/event/src/main.rs index 6d8bb472e..09d91418e 100644 --- a/services/event/src/main.rs +++ b/services/event/src/main.rs @@ -28,7 +28,11 @@ struct AppState { } async fn process_pending(state: &AppState) { - let conn = state.pool.get_conn().await; + let conn = state + .pool + .get_conn() + .await + .expect("failed to get db connection"); let rows = match conn .0 .query( diff --git a/services/graphql/src/main.rs b/services/graphql/src/main.rs index fcdc11210..1a07b4e12 100644 --- a/services/graphql/src/main.rs +++ b/services/graphql/src/main.rs @@ -19,7 +19,7 @@ use futures_util::{stream::Stream, StreamExt}; use httpclient::HttpClient as Client; use serde_json::json; use shutdown::shutdown_signal; -use std::{env, net::ToSocketAddrs, result::Result}; +use std::{env, net::ToSocketAddrs}; use tokio::net::TcpListener; use tower_http::cors::CorsLayer; use tungstenite::error::Error as WsError; @@ -36,13 +36,16 @@ impl Query { ctx: &Context<'_>, #[graphql(name = "account_name")] account_name: String, #[graphql(name = "auth_account")] auth_account: String, - ) -> String { - let account_from_token = get_auth_account(ctx, auth_account).unwrap(); + ) -> Result { + let account_from_token = get_auth_account(ctx, auth_account)?; let uri = Uri::new_from_env_var("BALANCE_BY_ACCOUNT_URL").to_string(); let body = account_auth(account_name, account_from_token); let client = Client::new(); - let response = client.post(uri, body).await.unwrap(); - response.text().await.unwrap() + let response_body = client + .post(uri, body) + .await + .map_err(|e| Error::new(e.to_string()))?; + Ok(response_body) } #[graphql(name = "transactionsByAccount")] @@ -51,15 +54,18 @@ impl Query { ctx: &Context<'_>, #[graphql(name = "account_name")] account_name: String, #[graphql(name = "auth_account")] auth_account: String, - ) -> Vec { - let account_from_token = get_auth_account(ctx, auth_account).unwrap(); + ) -> Result, Error> { + let account_from_token = get_auth_account(ctx, auth_account)?; let uri = Uri::new_from_env_var("TRANSACTIONS_BY_ACCOUNT_URL").to_string(); let body = account_auth(account_name, account_from_token); let client = Client::new(); - let response = client.post(uri, body).await.unwrap(); - let response_body = response.text().await.unwrap(); - let response_body: IntraTransactions = serde_json::from_str(&response_body).unwrap(); - response_body.transactions.0 + let response_body = client + .post(uri, body) + .await + .map_err(|e| Error::new(e.to_string()))?; + let parsed: IntraTransactions = serde_json::from_str(&response_body) + .map_err(|e| Error::new(format!("failed to parse response: {e}")))?; + Ok(parsed.transactions.0) } #[graphql(name = "transactionByID")] @@ -69,15 +75,18 @@ impl Query { #[graphql(name = "id")] id: String, #[graphql(name = "account_name")] account_name: String, #[graphql(name = "auth_account")] auth_account: String, - ) -> Transaction { - let account_from_token = get_auth_account(ctx, auth_account).unwrap(); + ) -> Result { + let account_from_token = get_auth_account(ctx, auth_account)?; let uri = Uri::new_from_env_var("TRANSACTION_BY_ID_URL").to_string(); let body = id_account_auth(id, account_name, account_from_token); let client = Client::new(); - let response = client.post(uri, body).await.unwrap(); - let response_body = response.text().await.unwrap(); - let intra_transaction: IntraTransaction = serde_json::from_str(&response_body).unwrap(); - intra_transaction.transaction + let response_body = client + .post(uri, body) + .await + .map_err(|e| Error::new(e.to_string()))?; + let intra_transaction: IntraTransaction = serde_json::from_str(&response_body) + .map_err(|e| Error::new(format!("failed to parse response: {e}")))?; + Ok(intra_transaction.transaction) } #[graphql(name = "requestsByAccount")] @@ -86,15 +95,18 @@ impl Query { ctx: &Context<'_>, #[graphql(name = "account_name")] account_name: String, #[graphql(name = "auth_account")] auth_account: String, - ) -> Vec { - let account_from_token = get_auth_account(ctx, auth_account).unwrap(); + ) -> Result, Error> { + let account_from_token = get_auth_account(ctx, auth_account)?; let uri = Uri::new_from_env_var("REQUESTS_BY_ACCOUNT_URL").to_string(); - let client = Client::new(); let body = account_auth(account_name, account_from_token); - let response = client.post(uri, body).await.unwrap(); - let response_body = response.text().await.unwrap(); - let response_body: IntraTransactions = serde_json::from_str(&response_body).unwrap(); - response_body.transactions.0 + let client = Client::new(); + let response_body = client + .post(uri, body) + .await + .map_err(|e| Error::new(e.to_string()))?; + let parsed: IntraTransactions = serde_json::from_str(&response_body) + .map_err(|e| Error::new(format!("failed to parse response: {e}")))?; + Ok(parsed.transactions.0) } #[graphql(name = "requestByID")] @@ -104,29 +116,35 @@ impl Query { #[graphql(name = "id")] id: String, #[graphql(name = "account_name")] account_name: String, #[graphql(name = "auth_account")] auth_account: String, - ) -> Transaction { - let account_from_token = get_auth_account(ctx, auth_account).unwrap(); + ) -> Result { + let account_from_token = get_auth_account(ctx, auth_account)?; let uri = Uri::new_from_env_var("REQUEST_BY_ID_URL").to_string(); let body = id_account_auth(id, account_name, account_from_token); let client = Client::new(); - let response = client.post(uri, body).await.unwrap(); - let response_body = response.text().await.unwrap(); - let intra_transaction: IntraTransaction = serde_json::from_str(&response_body).unwrap(); - intra_transaction.transaction + let response_body = client + .post(uri, body) + .await + .map_err(|e| Error::new(e.to_string()))?; + let intra_transaction: IntraTransaction = serde_json::from_str(&response_body) + .map_err(|e| Error::new(format!("failed to parse response: {e}")))?; + Ok(intra_transaction.transaction) } #[graphql(name = "rules")] async fn rules( &self, #[graphql(name = "transaction")] transaction: Transaction, - ) -> Transaction { + ) -> Result { let uri = Uri::new_from_env_var("RULE_URL").to_string(); - let client = Client::new(); let body = json!(transaction).to_string(); - let response = client.post(uri, body).await.unwrap(); - let response_body = response.text().await.unwrap(); - let intra_transaction: IntraTransaction = serde_json::from_str(&response_body).unwrap(); - intra_transaction.transaction + let client = Client::new(); + let response_body = client + .post(uri, body) + .await + .map_err(|e| Error::new(e.to_string()))?; + let intra_transaction: IntraTransaction = serde_json::from_str(&response_body) + .map_err(|e| Error::new(format!("failed to parse response: {e}")))?; + Ok(intra_transaction.transaction) } } @@ -139,16 +157,19 @@ impl Mutation { ctx: &Context<'_>, #[graphql(name = "transaction")] transaction: Transaction, #[graphql(name = "auth_account")] auth_account: String, - ) -> Transaction { - let account_from_token = get_auth_account(ctx, auth_account).unwrap(); + ) -> Result { + let account_from_token = get_auth_account(ctx, auth_account)?; let uri = Uri::new_from_env_var("REQUEST_CREATE_URL").to_string(); - let client = Client::new(); let request = IntraTransaction::new(account_from_token, transaction); let body = json!(request).to_string(); - let response = client.post(uri, body).await.unwrap(); - let response_body = response.text().await.unwrap(); - let intra_transaction: IntraTransaction = serde_json::from_str(&response_body).unwrap(); - intra_transaction.transaction + let client = Client::new(); + let response_body = client + .post(uri, body) + .await + .map_err(|e| Error::new(e.to_string()))?; + let intra_transaction: IntraTransaction = serde_json::from_str(&response_body) + .map_err(|e| Error::new(format!("failed to parse response: {e}")))?; + Ok(intra_transaction.transaction) } async fn approve_request( @@ -158,10 +179,9 @@ impl Mutation { #[graphql(name = "account_name")] account_name: String, #[graphql(name = "account_role")] account_role: AccountRole, #[graphql(name = "auth_account")] auth_account: String, - ) -> Transaction { - let account_from_token = get_auth_account(ctx, auth_account).unwrap(); + ) -> Result { + let account_from_token = get_auth_account(ctx, auth_account)?; let uri = Uri::new_from_env_var("REQUEST_APPROVE_URL").to_string(); - let client = Client::new(); let request = RequestApprove::new( account_from_token, transaction_id, @@ -169,10 +189,14 @@ impl Mutation { account_role, ); let body = json!(request).to_string(); - let response = client.post(uri, body).await.unwrap(); - let response_body = response.text().await.unwrap(); - let intra_transaction: IntraTransaction = serde_json::from_str(&response_body).unwrap(); - intra_transaction.transaction + let client = Client::new(); + let response_body = client + .post(uri, body) + .await + .map_err(|e| Error::new(e.to_string()))?; + let intra_transaction: IntraTransaction = serde_json::from_str(&response_body) + .map_err(|e| Error::new(format!("failed to parse response: {e}")))?; + Ok(intra_transaction.transaction) } } @@ -188,13 +212,20 @@ impl Subscription { #[graphql(name = "region")] region: Option, #[graphql(name = "municipality")] municipality: Option, ) -> impl Stream { - let resource = env::var("MEASURE_RESOURCE").unwrap(); - let uri = Uri::new_from_env_var("MEASURE_URL") - .with_path(resource.as_str()) - .with_ws() - .to_string(); - let ws_client = WsClient::new(uri, "gdp".to_string(), date, country, region, municipality); stream! { + let resource = match envvar::required("MEASURE_RESOURCE") { + Ok(r) => r, + Err(e) => { + tracing::error!("{}", e); + return; + } + }; + let uri = Uri::new_from_env_var("MEASURE_URL") + .with_path(resource.as_str()) + .with_ws() + .to_string(); + let ws_client = WsClient::new(uri, "gdp".to_string(), date, country, region, municipality); + let measure_socket = match ws_client.connect().await { Ok(ws) => { tracing::info!("graphql websocket connection created with measure"); @@ -218,9 +249,15 @@ impl Subscription { Some(Ok(msg)) => { match msg { tungstenite::Message::Text(text) => { - let gdp: f64 = serde_json::from_str(&text).unwrap(); - tracing::info!("sending gdp from measure: {}", gdp); - yield gdp; + match serde_json::from_str::(&text) { + Ok(gdp) => { + tracing::info!("sending gdp from measure: {}", gdp); + yield gdp; + } + Err(e) => { + tracing::error!("failed to parse gdp value from measure: {}", e); + } + } } _ => { tracing::info!("received non-text message from measure: {:?}", msg); @@ -267,36 +304,38 @@ fn id_account_auth(id: String, account_name: String, auth_account: String) -> St fn get_auth_account(ctx: &Context<'_>, account_from_request: String) -> Result { if env::var("ENABLE_API_AUTH") == Ok("true".to_string()) { - let headers = ctx.data::().unwrap(); - let amzn_ctx = get_amzn_ctx_from_headers(headers); - match amzn_ctx.authorizer { - Some(authorizer) => Ok(authorizer - .jwt - .unwrap() - .claims - .get("cognito:username") - .unwrap() - .to_string()), - None => { - tracing::error!("error: missing authorizer"); - Err("error: missing authorizer".into()) - } - } + let headers = ctx.data::()?; + let amzn_ctx = get_amzn_ctx_from_headers(headers)?; + let authorizer = amzn_ctx + .authorizer + .ok_or_else(|| Error::new("missing authorizer"))?; + let jwt = authorizer + .jwt + .ok_or_else(|| Error::new("missing jwt in authorizer"))?; + let username = jwt + .claims + .get("cognito:username") + .ok_or_else(|| Error::new("missing cognito:username claim"))?; + Ok(username.to_string()) } else { Ok(account_from_request) } } -fn get_amzn_ctx_from_headers(headers: &HeaderMap) -> ApiGatewayV2httpRequestContext { - headers +fn get_amzn_ctx_from_headers(headers: &HeaderMap) -> Result { + let value = headers .get("x-amzn-request-context") - .and_then(|value| serde_json::from_str(value.to_str().unwrap()).ok()) - .unwrap() + .ok_or_else(|| Error::new("missing x-amzn-request-context header"))?; + let value_str = value + .to_str() + .map_err(|e| Error::new(format!("invalid x-amzn-request-context header: {e}")))?; + serde_json::from_str(value_str) + .map_err(|e| Error::new(format!("failed to parse x-amzn-request-context: {e}"))) } async fn graphiql() -> impl IntoResponse { - let graphql_resource = std::env::var("GRAPHQL_RESOURCE").unwrap(); - let graphql_ws_resource = std::env::var("GRAPHQL_WS_RESOURCE").unwrap(); + let graphql_resource = envvar::required("GRAPHQL_RESOURCE").unwrap(); + let graphql_ws_resource = envvar::required("GRAPHQL_WS_RESOURCE").unwrap(); response::Html( http::GraphiQLSource::build() .endpoint(format!("/{graphql_resource}").as_str()) @@ -340,13 +379,12 @@ async fn main() { tracing_subscriber::fmt().init(); } - let readiness_check_path = env::var(READINESS_CHECK_PATH) - .unwrap_or_else(|_| panic!("{READINESS_CHECK_PATH} variable assignment")); + let readiness_check_path = envvar::required(READINESS_CHECK_PATH).unwrap(); let schema = Schema::build(Query, Mutation, Subscription).finish(); - let graphql_resource = std::env::var("GRAPHQL_RESOURCE").unwrap(); - let graphql_ws_resource = std::env::var("GRAPHQL_WS_RESOURCE").unwrap(); + let graphql_resource = envvar::required("GRAPHQL_RESOURCE").unwrap(); + let graphql_ws_resource = envvar::required("GRAPHQL_WS_RESOURCE").unwrap(); let app = Router::new() .route("/", get(graphiql)) @@ -362,9 +400,9 @@ async fn main() { .layer(CorsLayer::permissive()) .with_state(schema); - let hostname_or_ip = env::var("HOSTNAME_OR_IP").unwrap_or("0.0.0.0".to_string()); + let hostname_or_ip = envvar::optional("HOSTNAME_OR_IP", "0.0.0.0"); - let port = env::var("GRAPHQL_PORT").unwrap(); + let port = envvar::required("GRAPHQL_PORT").unwrap(); let serve_addr = format!("{hostname_or_ip}:{port}"); diff --git a/services/measure/src/main.rs b/services/measure/src/main.rs index 5f723534a..a9d869d8e 100644 --- a/services/measure/src/main.rs +++ b/services/measure/src/main.rs @@ -59,8 +59,7 @@ async fn main() { tracing_subscriber::fmt().init(); } - let readiness_check_path = env::var(READINESS_CHECK_PATH) - .unwrap_or_else(|_| panic!("{READINESS_CHECK_PATH} variable assignment")); + let readiness_check_path = envvar::required(READINESS_CHECK_PATH).unwrap(); let conn_uri = DB::create_conn_uri_from_env_vars(); let pool = DB::new_pool(&conn_uri).await; @@ -73,8 +72,8 @@ async fn main() { ) .with_state(pool); - let hostname_or_ip = env::var("HOSTNAME_OR_IP").unwrap_or("0.0.0.0".to_string()); - let port = env::var("MEASURE_PORT").unwrap(); + let hostname_or_ip = envvar::optional("HOSTNAME_OR_IP", "0.0.0.0"); + let port = envvar::required("MEASURE_PORT").unwrap(); let serve_addr = format!("{hostname_or_ip}:{port}"); let listener = tokio::net::TcpListener::bind(serve_addr.clone()) @@ -104,7 +103,7 @@ async fn ws_handler( /// spawn websocket per connection async fn handle_socket(socket: WebSocket, _who: SocketAddr, pool: ConnectionPool, params: Params) { // get abbreviated location names from postgres - let conn = pool.get_conn().await; + let conn = pool.get_conn().await.expect("failed to get db connection"); let abbreviations = abbrev_names(&conn, params).await; // concat abbreviated location names to colon separated key: 2024-08-20:gdp:usa:cal:sac diff --git a/services/request-approve/src/main.rs b/services/request-approve/src/main.rs index 1851c78af..dad4d54ed 100644 --- a/services/request-approve/src/main.rs +++ b/services/request-approve/src/main.rs @@ -5,9 +5,9 @@ use axum::{ Router, }; use pg::postgres::{ConnectionPool, DB}; -use service::Service; +use service::{Service, ServiceError}; use shutdown::shutdown_signal; -use std::{env, net::ToSocketAddrs}; +use std::net::ToSocketAddrs; use tokio::net::TcpListener; use types::request_response::{IntraTransaction, RequestApprove}; @@ -17,26 +17,28 @@ const READINESS_CHECK_PATH: &str = "READINESS_CHECK_PATH"; async fn handle_event( State(pool): State, event: Json, -) -> Result, StatusCode> { +) -> Result, ServiceError> { let client_request = event.0; - let conn = pool.get_conn().await; + let conn = pool + .get_conn() + .await + .map_err(|_| ServiceError::internal("failed to get db connection"))?; let svc = Service::new(&conn, None); - let request_id = client_request.id.parse::().unwrap(); + let request_id = client_request + .id + .parse::() + .map_err(|e| ServiceError::bad_request(&format!("invalid request id: {e}")))?; let transaction_request = svc .get_full_transaction_by_id(request_id) .await - .map_err(|e| { - tracing::error!("error: {}", e); - StatusCode::INTERNAL_SERVER_ERROR - })?; + .map_err(|e| ServiceError::internal(&e.to_string()))?; if transaction_request.equilibrium_time.is_some() { - println!("transaction previously approved"); - return Err(StatusCode::BAD_REQUEST); + return Err(ServiceError::bad_request("transaction previously approved")); } let auth_account = client_request.auth_account; @@ -45,7 +47,7 @@ async fn handle_event( let approved_transaction_request = svc .approve(auth_account, approver_role, transaction_request) .await - .unwrap(); // todo: handle error + .map_err(|e| ServiceError::internal(&e.to_string()))?; Ok(axum::Json(approved_transaction_request)) } @@ -54,8 +56,7 @@ async fn handle_event( async fn main() { tracing_subscriber::fmt::init(); - let readiness_check_path = env::var(READINESS_CHECK_PATH) - .unwrap_or_else(|_| panic!("{READINESS_CHECK_PATH} variable assignment")); + let readiness_check_path = envvar::required(READINESS_CHECK_PATH).unwrap(); let conn_uri = DB::create_conn_uri_from_env_vars(); @@ -69,9 +70,9 @@ async fn main() { ) .with_state(pool); - let hostname_or_ip = env::var("HOSTNAME_OR_IP").unwrap_or("0.0.0.0".to_string()); + let hostname_or_ip = envvar::optional("HOSTNAME_OR_IP", "0.0.0.0"); - let port = env::var("REQUEST_APPROVE_PORT").unwrap(); + let port = envvar::required("REQUEST_APPROVE_PORT").unwrap(); let serve_addr = format!("{hostname_or_ip}:{port}"); diff --git a/services/request-by-id/src/main.rs b/services/request-by-id/src/main.rs index 960ddc2f0..818e08105 100644 --- a/services/request-by-id/src/main.rs +++ b/services/request-by-id/src/main.rs @@ -7,7 +7,7 @@ use axum::{ use pg::postgres::{ConnectionPool, DB}; use service::Service; use shutdown::shutdown_signal; -use std::{env, net::ToSocketAddrs}; +use std::net::ToSocketAddrs; use tokio::net::TcpListener; use types::request_response::{IntraTransaction, QueryById}; @@ -20,7 +20,10 @@ async fn handle_event( ) -> Result, StatusCode> { let client_request = event.0; - let conn = pool.get_conn().await; + let conn = pool + .get_conn() + .await + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; let svc = Service::new(&conn, None); @@ -73,8 +76,7 @@ async fn handle_event( async fn main() { tracing_subscriber::fmt::init(); - let readiness_check_path = env::var(READINESS_CHECK_PATH) - .unwrap_or_else(|_| panic!("{READINESS_CHECK_PATH} variable assignment")); + let readiness_check_path = envvar::required(READINESS_CHECK_PATH).unwrap(); let conn_uri = DB::create_conn_uri_from_env_vars(); @@ -88,9 +90,9 @@ async fn main() { ) .with_state(pool); - let hostname_or_ip = env::var("HOSTNAME_OR_IP").unwrap_or("0.0.0.0".to_string()); + let hostname_or_ip = envvar::optional("HOSTNAME_OR_IP", "0.0.0.0"); - let port = env::var("REQUEST_BY_ID_PORT").unwrap(); + let port = envvar::required("REQUEST_BY_ID_PORT").unwrap(); let serve_addr = format!("{hostname_or_ip}:{port}"); diff --git a/services/request-create/src/main.rs b/services/request-create/src/main.rs index e16558746..ccdfa93da 100644 --- a/services/request-create/src/main.rs +++ b/services/request-create/src/main.rs @@ -7,9 +7,9 @@ use axum::{ use cache::Cache; use httpclient::HttpClient as Client; use pg::postgres::{ConnectionPool, DatabaseConnection, DB}; -use service::Service; +use service::{Service, ServiceError}; use shutdown::shutdown_signal; -use std::{env, error::Error, net::ToSocketAddrs, sync::Arc}; +use std::{error::Error, net::ToSocketAddrs, sync::Arc}; use thiserror::Error; use tokio::net::TcpListener; use types::{request_response::IntraTransaction, transaction::Transaction}; @@ -42,8 +42,7 @@ pub async fn get_rule_applied_transaction( let uri = Uri::new_from_env_var("RULE_URL").to_string(); let client = Client::new(); let body = serde_json::to_string(&transaction)?; - let response = client.post(uri, body).await?; - let rule_tested = response.text().await.unwrap(); + let rule_tested = client.post(uri, body).await?; let intra_transaction = IntraTransaction::from_json_string(rule_tested.as_str())?; Ok(intra_transaction) } @@ -86,22 +85,12 @@ async fn test_values(req: IntraTransaction) -> Result { - // println!("client request equal to rule response") - } - Err(e) => { - return Err(Box::new(e)); - } - } + .test_equality(req.clone().transaction.transaction_items)?; let mut response = req.clone(); response.add_rule_tested_values(rule_tested); @@ -112,13 +101,10 @@ async fn test_values(req: IntraTransaction) -> Result, -) -> Result { +) -> Result> { let accounts = rule_tested.transaction.transaction_items.list_accounts(); - let profile_ids = svc - .get_profile_ids_by_account_names(accounts) - .await - .unwrap(); + let profile_ids = svc.get_profile_ids_by_account_names(accounts).await?; let mut transaction_request = rule_tested.transaction.clone(); @@ -126,11 +112,13 @@ async fn create_request( .transaction_items .add_profile_ids(profile_ids); - let transaction_id = svc.create_transaction(transaction_request).await.unwrap(); + let transaction_id = svc.create_transaction(transaction_request).await?; - let id = transaction_id.parse::().unwrap(); + let id = transaction_id + .parse::() + .map_err(|e| format!("invalid transaction id: {e}"))?; - let inserted_transaction_request = svc.get_full_transaction_by_id(id).await.unwrap(); + let inserted_transaction_request = svc.get_full_transaction_by_id(id).await?; Ok(inserted_transaction_request) } @@ -138,37 +126,47 @@ async fn create_request( async fn handle_event( State(store): State, intra_transaction: Json, -) -> Result, StatusCode> { +) -> Result, ServiceError> { let request = intra_transaction.0; let rule_tested = test_values(request) .await - .map_err(|_e| { - println!("error: {_e:?}"); - StatusCode::INTERNAL_SERVER_ERROR - }) - .unwrap(); + .map_err(|e| ServiceError::bad_request(&e.to_string()))?; - let conn = store.pool.get_conn().await; + let conn = store + .pool + .get_conn() + .await + .map_err(|_| ServiceError::internal("failed to get db connection"))?; let svc = Service::new(&conn, store.cache.clone()); let inserted_transaction_request = create_request(rule_tested.clone(), &svc) .await - .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; + .map_err(|e| ServiceError::internal(&e.to_string()))?; // auth_account and approver_role are different on request-approve - let auth_account = rule_tested.clone().auth_account.unwrap(); - let approver_role = rule_tested.transaction.author_role.unwrap(); + let auth_account = rule_tested + .clone() + .auth_account + .ok_or_else(|| ServiceError::bad_request("missing auth account"))?; + let approver_role = rule_tested + .transaction + .author_role + .ok_or_else(|| ServiceError::bad_request("missing author role"))?; - let conn = store.pool.get_conn().await; + let conn = store + .pool + .get_conn() + .await + .map_err(|_| ServiceError::internal("failed to get db connection"))?; let svc = Service::new(&conn, store.cache.clone()); let approved_transaction_request = svc .approve(auth_account, approver_role, inserted_transaction_request) .await - .unwrap(); // todo: handle error + .map_err(|e| ServiceError::internal(&e.to_string()))?; Ok(axum::Json(approved_transaction_request)) } @@ -177,8 +175,7 @@ async fn handle_event( async fn main() { tracing_subscriber::fmt::init(); - let readiness_check_path = env::var(READINESS_CHECK_PATH) - .unwrap_or_else(|_| panic!("{READINESS_CHECK_PATH} variable assignment")); + let readiness_check_path = envvar::required(READINESS_CHECK_PATH).unwrap(); let conn_uri = DB::create_conn_uri_from_env_vars(); @@ -196,9 +193,9 @@ async fn main() { ) .with_state(store); - let hostname_or_ip = env::var("HOSTNAME_OR_IP").unwrap_or("0.0.0.0".to_string()); + let hostname_or_ip = envvar::optional("HOSTNAME_OR_IP", "0.0.0.0"); - let port = env::var("REQUEST_CREATE_PORT").unwrap(); + let port = envvar::required("REQUEST_CREATE_PORT").unwrap(); let serve_addr = format!("{hostname_or_ip}:{port}"); diff --git a/services/requests-by-account/src/main.rs b/services/requests-by-account/src/main.rs index 50730bca2..e05e6909f 100644 --- a/services/requests-by-account/src/main.rs +++ b/services/requests-by-account/src/main.rs @@ -7,7 +7,7 @@ use axum::{ use pg::postgres::{ConnectionPool, DB}; use service::Service; use shutdown::shutdown_signal; -use std::{env, net::ToSocketAddrs}; +use std::net::ToSocketAddrs; use tokio::net::TcpListener; use types::request_response::{IntraTransactions, QueryByAccount}; @@ -20,14 +20,17 @@ async fn handle_event( ) -> Result, StatusCode> { let client_request = event.0; - let conn = pool.get_conn().await; + let conn = pool + .get_conn() + .await + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; let svc = Service::new(&conn, None); let account = client_request.account_name; - let record_limit = env::var("RETURN_RECORD_LIMIT") - .unwrap_or_else(|_| panic!("RETURN_RECORD_LIMIT variable assignment")) + let record_limit = envvar::required("RETURN_RECORD_LIMIT") + .unwrap() .parse::() .unwrap(); @@ -52,8 +55,7 @@ async fn handle_event( async fn main() { tracing_subscriber::fmt::init(); - let readiness_check_path = env::var(READINESS_CHECK_PATH) - .unwrap_or_else(|_| panic!("{READINESS_CHECK_PATH} variable assignment")); + let readiness_check_path = envvar::required(READINESS_CHECK_PATH).unwrap(); let conn_uri = DB::create_conn_uri_from_env_vars(); @@ -67,9 +69,9 @@ async fn main() { ) .with_state(pool); - let hostname_or_ip = env::var("HOSTNAME_OR_IP").unwrap_or("0.0.0.0".to_string()); + let hostname_or_ip = envvar::optional("HOSTNAME_OR_IP", "0.0.0.0"); - let port = env::var("REQUESTS_BY_ACCOUNT_PORT").unwrap(); + let port = envvar::required("REQUESTS_BY_ACCOUNT_PORT").unwrap(); let serve_addr = format!("{hostname_or_ip}:{port}"); diff --git a/services/rule/src/lib.rs b/services/rule/src/lib.rs index d4c36aca8..b0196be4b 100644 --- a/services/rule/src/lib.rs +++ b/services/rule/src/lib.rs @@ -37,21 +37,20 @@ pub fn create_response( pub fn label_approved_transaction_items( role_sequence: &RoleSequence, transaction_items: TransactionItems, -) -> TransactionItems { +) -> Result { let mut labeled = transaction_items; for tr_item in labeled.0.iter_mut() { for role in role_sequence { // get list of approvals per role (debitor or creditor) - let approvals_per_role = tr_item + let approvals = tr_item .approvals .clone() - .unwrap() - .get_approvals_per_role(*role); + .ok_or("missing approvals in transaction item")?; + let approvals_per_role = approvals.get_approvals_per_role(*role); - // todo: error on 0 approvals_per_role if approvals_per_role.0.is_empty() { - panic!("0 approvals per role"); + return Err("0 approvals per role".into()); } let mut approval_count = 0; @@ -94,7 +93,7 @@ pub fn label_approved_transaction_items( } } } - labeled + Ok(labeled) } #[cfg(test)] @@ -190,7 +189,8 @@ mod tests { }]); // test function - let got = label_approved_transaction_items(&DEBITOR_FIRST, tr_items); + let got = label_approved_transaction_items(&DEBITOR_FIRST, tr_items) + .expect("label_approved_transaction_items returned an error"); // assert #1 // save creditor_approval_time diff --git a/services/rule/src/main.rs b/services/rule/src/main.rs index 4f2971844..ff779d8cb 100644 --- a/services/rule/src/main.rs +++ b/services/rule/src/main.rs @@ -7,9 +7,9 @@ use axum::{ use cache::Cache; use pg::postgres::{ConnectionPool, DatabaseConnection, DB}; use rule::{create_response, expected_values, label_approved_transaction_items}; -use service::Service; +use service::{Service, ServiceError}; use shutdown::shutdown_signal; -use std::{env, net::ToSocketAddrs, sync::Arc}; +use std::{net::ToSocketAddrs, sync::Arc}; use types::approval::{Approval, Approvals}; use types::{ account_role::{RoleSequence, CREDITOR_FIRST, DEBITOR_FIRST}, @@ -33,10 +33,13 @@ async fn apply_transaction_item_rules<'a>( svc: &'a Service<'a, DatabaseConnection>, role_sequence: RoleSequence, transaction_items: &TransactionItems, -) -> TransactionItems { +) -> Result { let accounts = transaction_items.list_accounts(); - let initial_account_profiles = svc.get_account_profiles(accounts).await.unwrap(); + let initial_account_profiles = svc + .get_account_profiles(accounts) + .await + .map_err(|e| ServiceError::internal(&e.to_string()))?; let mut response: TransactionItems = TransactionItems::default(); @@ -54,16 +57,21 @@ async fn apply_transaction_item_rules<'a>( // get account profile let account_profile = initial_account_profiles .match_profile_by_account(account.clone()) - .unwrap(); + .ok_or_else(|| { + ServiceError::internal(&format!("profile not found for {account}")) + })?; // add profile id to transaction item - current_tr_item.set_profile_id(role, account_profile.clone().id.unwrap()); // todo: handle missing id error + let profile_id = account_profile.clone().id.ok_or_else(|| { + ServiceError::internal(&format!("missing profile id for {account}")) + })?; + current_tr_item.set_profile_id(role, profile_id); // get rules matching state in account profile let state_rules = svc .get_state_tr_item_rule_instances(role, account_profile.state_name.to_string()) .await - .unwrap(); + .map_err(|e| ServiceError::internal(&e.to_string()))?; // apply state rules to transaction item for rule_instance in state_rules.clone().0.iter() { @@ -71,7 +79,7 @@ async fn apply_transaction_item_rules<'a>( rule_instance, current_tr_item.clone(), ) - .unwrap(); + .map_err(|e| ServiceError::internal(&e.to_string()))?; // first item is updated original, rest are computed if !state_added_tr_items.0.is_empty() { current_tr_item = state_added_tr_items.0[0].clone(); @@ -85,7 +93,7 @@ async fn apply_transaction_item_rules<'a>( let account_rules = svc .get_tr_item_rule_instances_by_role_account(role, account.clone()) .await - .unwrap(); + .map_err(|e| ServiceError::internal(&e.to_string()))?; // apply account rules to transaction item for rule_instance in account_rules.clone().0.iter() { @@ -93,7 +101,7 @@ async fn apply_transaction_item_rules<'a>( rule_instance, current_tr_item.clone(), ) - .unwrap(); + .map_err(|e| ServiceError::internal(&e.to_string()))?; // first item is updated original, rest are computed if !account_added_tr_items.0.is_empty() { current_tr_item = account_added_tr_items.0[0].clone(); @@ -112,7 +120,7 @@ async fn apply_transaction_item_rules<'a>( let added_profile_ids = svc .get_profile_ids_by_account_names(added_accounts) .await - .unwrap(); + .map_err(|e| ServiceError::internal(&e.to_string()))?; // add account profile ids to rule added transaction items // todo: some profiles may be previously fetched when @@ -127,7 +135,7 @@ async fn apply_transaction_item_rules<'a>( response.0.append(&mut rule_added.0); } - response + Ok(response) } async fn apply_approval_rules<'a>( @@ -135,7 +143,7 @@ async fn apply_approval_rules<'a>( role_sequence: RoleSequence, transaction_items: &mut TransactionItems, approval_time: &TZTime, -) { +) -> Result<(), ServiceError> { // loop through transaction_item(s) for tr_item in transaction_items.0.iter_mut() { // create empty list of approvals which will increase as @@ -149,7 +157,10 @@ async fn apply_approval_rules<'a>( let account = tr_item.get_account_by_role(role); // query account owners (approvers) of debitor or credit account - let approvers = svc.get_account_approvers(account).await.unwrap(); + let approvers = svc + .get_account_approvers(account) + .await + .map_err(|e| ServiceError::internal(&e.to_string()))?; // loop through list of approvers for approver in approvers { @@ -172,7 +183,7 @@ async fn apply_approval_rules<'a>( let approval_rules = svc .get_approval_rule_instances(role, approver.clone()) .await - .unwrap(); + .map_err(|e| ServiceError::internal(&e.to_string()))?; // loop through each approval rule and apply for rule_instance in approval_rules.0.iter() { @@ -183,7 +194,7 @@ async fn apply_approval_rules<'a>( &mut approval, approval_time, ) - .unwrap(); // todo: handle error + .map_err(|e| ServiceError::internal(&e.to_string()))?; } // add post rule approval to approvals list @@ -194,6 +205,8 @@ async fn apply_approval_rules<'a>( // attach post rule approvals to each transaction_item tr_item.approvals = Some(approvals); } + + Ok(()) } // build transaction items from transaction_item_rule_instance templates @@ -201,10 +214,10 @@ async fn apply_approval_rules<'a>( async fn build_items_from_rule_instance( conn: &DatabaseConnection, transaction_rule_instance_id: &str, -) -> Result { +) -> Result { let tri_id: i32 = transaction_rule_instance_id .parse() - .map_err(|_| StatusCode::BAD_REQUEST)?; + .map_err(|_| ServiceError::bad_request("invalid rule_instance_id"))?; let rows = conn .0 .query( @@ -216,7 +229,7 @@ async fn build_items_from_rule_instance( .await .map_err(|e| { tracing::error!("query transaction_item_rule_instance failed: {}", e); - StatusCode::INTERNAL_SERVER_ERROR + ServiceError::internal("query transaction_item_rule_instance failed") })?; let mut items = TransactionItems::default(); @@ -256,9 +269,11 @@ async fn build_items_from_rule_instance( async fn apply_rules( State(state): State, transaction: Json, -) -> Result, StatusCode> { +) -> Result, ServiceError> { if !expected_values(&transaction) { - return Err(StatusCode::BAD_REQUEST); + return Err(ServiceError::bad_request( + "missing expected transaction values", + )); }; let debitor_first = transaction.debitor_first.unwrap_or(false); @@ -269,7 +284,11 @@ async fn apply_rules( }; // get connection from pool - let conn = state.pool.get_conn().await; + let conn = state + .pool + .get_conn() + .await + .map_err(|_| ServiceError::internal("failed to get db connection"))?; // when rule_instance_id is set and items are empty, build items from templates let mut transaction_items = if let Some(ref rule_instance_id) = transaction.rule_instance_id { @@ -289,7 +308,7 @@ async fn apply_rules( let svc = Service::new(&conn, state.cache.clone()); let mut rule_applied_tr_items = - apply_transaction_item_rules(&svc, role_sequence, &transaction_items).await; + apply_transaction_item_rules(&svc, role_sequence, &transaction_items).await?; // create an approval time to be used for all automated approvals let approval_time = TZTime::now(); @@ -300,9 +319,10 @@ async fn apply_rules( &mut rule_applied_tr_items, &approval_time, ) - .await; + .await?; - let labeled_approved = label_approved_transaction_items(&role_sequence, rule_applied_tr_items); + let labeled_approved = label_approved_transaction_items(&role_sequence, rule_applied_tr_items) + .map_err(|e| ServiceError::internal(&e))?; let response_transaction = create_response(labeled_approved, &transaction); @@ -315,8 +335,7 @@ async fn apply_rules( async fn main() { tracing_subscriber::fmt::init(); - let readiness_check_path = env::var(READINESS_CHECK_PATH) - .unwrap_or_else(|_| panic!("{READINESS_CHECK_PATH} variable assignment")); + let readiness_check_path = envvar::required(READINESS_CHECK_PATH).unwrap(); let conn_uri = DB::create_conn_uri_from_env_vars(); @@ -334,9 +353,9 @@ async fn main() { ) .with_state(store); - let hostname_or_ip = env::var("HOSTNAME_OR_IP").unwrap_or("0.0.0.0".to_string()); + let hostname_or_ip = envvar::optional("HOSTNAME_OR_IP", "0.0.0.0"); - let port = env::var("RULE_PORT").unwrap(); + let port = envvar::required("RULE_PORT").unwrap(); let serve_addr = format!("{hostname_or_ip}:{port}"); diff --git a/services/rule/src/rules/approval.rs b/services/rule/src/rules/approval.rs index ae90ec3ca..ccb4f2d2b 100644 --- a/services/rule/src/rules/approval.rs +++ b/services/rule/src/rules/approval.rs @@ -34,7 +34,10 @@ fn approve_item_between_accounts( let debitor = rule_instance.variable_values[0].clone(); let creditor = rule_instance.variable_values[1].clone(); let item_id = rule_instance.variable_values[2].clone(); - let approver_role: AccountRole = rule_instance.variable_values[3].clone().parse().unwrap(); + let approver_role: AccountRole = rule_instance.variable_values[3] + .clone() + .parse() + .map_err(|e| format!("unsupported approver role: {e}"))?; let approver_account = rule_instance.variable_values[4].clone(); // match debitor/creditor/item_id on transaction_item and approver on approval @@ -52,6 +55,8 @@ fn approve_item_between_accounts( Ok(()) } +// approveAnyCreditItem: auto-approves creditor items +// variable_values = [CREDITOR, APPROVER_ROLE, APPROVER_NAME] fn approve_any_credit_item( rule_instance: &ApprovalRuleInstance, transaction_item: &TransactionItem, @@ -59,7 +64,10 @@ fn approve_any_credit_item( approval_time: &TZTime, ) -> Result<(), Box> { let _creditor = rule_instance.variable_values[0].clone(); - let approver_role: AccountRole = rule_instance.variable_values[1].clone().parse().unwrap(); + let approver_role: AccountRole = rule_instance.variable_values[1] + .clone() + .parse() + .map_err(|e| format!("unsupported approver role: {e}"))?; let approver_account = rule_instance.variable_values[2].clone(); let rule_instance_id = rule_instance.id.clone(); let transaction_id = transaction_item.transaction_id.clone(); @@ -77,3 +85,110 @@ fn approve_any_credit_item( Err("unmatched approver rule instance from db".into()) } } + +#[cfg(test)] +mod tests { + use super::*; + + fn test_approval_rule_instance( + rule_name: &str, + variable_values: Vec<&str>, + ) -> ApprovalRuleInstance { + ApprovalRuleInstance { + id: Some("1".to_string()), + rule_name: rule_name.to_string(), + rule_instance_name: "TestRule".to_string(), + variable_values: variable_values.iter().map(|s| s.to_string()).collect(), + account_role: AccountRole::Creditor, + account_name: "GroceryStore".to_string(), + disabled_time: None, + removed_time: None, + created_at: None, + } + } + + fn test_transaction_item() -> TransactionItem { + TransactionItem { + id: None, + transaction_id: None, + item_id: "bread".to_string(), + price: "3.000".to_string(), + quantity: "2.000".to_string(), + rule_instance_id: None, + rule_exec_ids: Some(vec![]), + unit_of_measurement: None, + units_measured: None, + debitor: "JacobWebb".to_string(), + creditor: "GroceryStore".to_string(), + debitor_profile_id: None, + creditor_profile_id: None, + debitor_approval_time: None, + creditor_approval_time: None, + debitor_rejection_time: None, + creditor_rejection_time: None, + debitor_expiration_time: None, + creditor_expiration_time: None, + approvals: None, + } + } + + fn test_approval() -> Approval { + Approval { + id: None, + rule_instance_id: None, + transaction_id: None, + transaction_item_id: None, + account_name: "GroceryStore".to_string(), + account_role: AccountRole::Creditor, + device_id: None, + device_latlng: None, + approval_time: None, + rejection_time: None, + expiration_time: None, + } + } + + #[test] + fn it_errors_on_unsupported_role_in_approve_any_credit_item() { + let rule_instance = test_approval_rule_instance( + "approveAnyCreditItem", + vec!["GroceryStore", "not_a_role", "GroceryStore"], + ); + let tr_item = test_transaction_item(); + let mut approval = test_approval(); + let approval_time = TZTime::now(); + + let result = match_approval_rule(&rule_instance, &tr_item, &mut approval, &approval_time); + + assert!(result.is_err()); + assert!(result + .unwrap_err() + .to_string() + .contains("unsupported approver role")); + } + + #[test] + fn it_errors_on_unsupported_role_in_approve_item_between_accounts() { + let rule_instance = test_approval_rule_instance( + "approveItemBetweenAccounts", + vec![ + "JacobWebb", + "GroceryStore", + "bread", + "not_a_role", + "GroceryStore", + ], + ); + let tr_item = test_transaction_item(); + let mut approval = test_approval(); + let approval_time = TZTime::now(); + + let result = match_approval_rule(&rule_instance, &tr_item, &mut approval, &approval_time); + + assert!(result.is_err()); + assert!(result + .unwrap_err() + .to_string() + .contains("unsupported approver role")); + } +} diff --git a/services/rule/src/rules/transaction_item.rs b/services/rule/src/rules/transaction_item.rs index 7fa7d3724..d12689362 100644 --- a/services/rule/src/rules/transaction_item.rs +++ b/services/rule/src/rules/transaction_item.rs @@ -47,12 +47,24 @@ fn append_multiplied_item_value( let debitor = rule_instance.variable_values[0].clone(); let creditor = rule_instance.variable_values[1].clone(); let item_name = rule_instance.variable_values[2].clone(); - let factor: f32 = rule_instance.variable_values[3].clone().parse().unwrap(); + let factor: f32 = rule_instance.variable_values[3] + .clone() + .parse() + .map_err(|e| format!("unsupported factor: {e}"))?; - let price: f32 = transaction_item.price.clone().parse().unwrap(); - let quantity = - utils::number_to_fixed_string(transaction_item.quantity.clone().parse::().unwrap()); - let rule_instance_id = rule_instance.id.clone().unwrap(); + let price: f32 = transaction_item + .price + .clone() + .parse() + .map_err(|e| format!("unsupported price: {e}"))?; + let quantity = utils::number_to_fixed_string( + transaction_item + .quantity + .clone() + .parse::() + .map_err(|e| format!("unsupported quantity: {e}"))?, + ); + let rule_instance_id = rule_instance.id.clone().ok_or("missing rule_instance id")?; let rule_exec_id = utils::create_rule_exec_id(); // clone original and add rule_exec_id @@ -60,7 +72,7 @@ fn append_multiplied_item_value( original_with_exec_id .rule_exec_ids .as_mut() - .unwrap() + .ok_or("missing rule_exec_ids")? .push(rule_exec_id.clone()); let post_token_debitor: String = match debitor.as_str() { @@ -228,4 +240,55 @@ mod tests { assert!(result.is_err()); } + + #[test] + fn it_errors_on_non_numeric_factor() { + let rule_instance = test_rule_instance( + "appendMultipliedItemValue", + vec!["ANY", "StateOfCalifornia", "tax", "not_a_number"], + ); + let item = test_transaction_item("10.000", "1"); + + let result = append_multiplied_item_value(&rule_instance, item); + + assert!(result.is_err()); + assert!(result + .unwrap_err() + .to_string() + .contains("unsupported factor")); + } + + #[test] + fn it_errors_on_non_numeric_price() { + let rule_instance = test_rule_instance( + "appendMultipliedItemValue", + vec!["ANY", "StateOfCalifornia", "tax", "0.09"], + ); + let item = test_transaction_item("not_a_number", "1"); + + let result = append_multiplied_item_value(&rule_instance, item); + + assert!(result.is_err()); + assert!(result + .unwrap_err() + .to_string() + .contains("unsupported price")); + } + + #[test] + fn it_errors_on_non_numeric_quantity() { + let rule_instance = test_rule_instance( + "appendMultipliedItemValue", + vec!["ANY", "StateOfCalifornia", "tax", "0.09"], + ); + let item = test_transaction_item("10.000", "not_a_number"); + + let result = append_multiplied_item_value(&rule_instance, item); + + assert!(result.is_err()); + assert!(result + .unwrap_err() + .to_string() + .contains("unsupported quantity")); + } } diff --git a/services/transaction-by-id/src/main.rs b/services/transaction-by-id/src/main.rs index 981951a2f..19b9ef2f1 100644 --- a/services/transaction-by-id/src/main.rs +++ b/services/transaction-by-id/src/main.rs @@ -7,7 +7,7 @@ use axum::{ use pg::postgres::{ConnectionPool, DB}; use service::Service; use shutdown::shutdown_signal; -use std::{env, net::ToSocketAddrs}; +use std::net::ToSocketAddrs; use tokio::net::TcpListener; use types::request_response::{IntraTransaction, QueryById}; @@ -20,7 +20,10 @@ async fn handle_event( ) -> Result, StatusCode> { let client_request = event.0; - let conn = pool.get_conn().await; + let conn = pool + .get_conn() + .await + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; let svc = Service::new(&conn, None); @@ -87,8 +90,7 @@ async fn handle_event( async fn main() { tracing_subscriber::fmt::init(); - let readiness_check_path = env::var(READINESS_CHECK_PATH) - .unwrap_or_else(|_| panic!("{READINESS_CHECK_PATH} variable assignment")); + let readiness_check_path = envvar::required(READINESS_CHECK_PATH).unwrap(); let conn_uri = DB::create_conn_uri_from_env_vars(); @@ -102,9 +104,9 @@ async fn main() { ) .with_state(pool); - let hostname_or_ip = env::var("HOSTNAME_OR_IP").unwrap_or("0.0.0.0".to_string()); + let hostname_or_ip = envvar::optional("HOSTNAME_OR_IP", "0.0.0.0"); - let port = env::var("TRANSACTION_BY_ID_PORT").unwrap(); + let port = envvar::required("TRANSACTION_BY_ID_PORT").unwrap(); let serve_addr = format!("{hostname_or_ip}:{port}"); diff --git a/services/transactions-by-account/src/main.rs b/services/transactions-by-account/src/main.rs index 4e04c93f9..29d9c25a9 100644 --- a/services/transactions-by-account/src/main.rs +++ b/services/transactions-by-account/src/main.rs @@ -7,7 +7,7 @@ use axum::{ use pg::postgres::{ConnectionPool, DB}; use service::Service; use shutdown::shutdown_signal; -use std::{env, net::ToSocketAddrs}; +use std::net::ToSocketAddrs; use tokio::net::TcpListener; use types::request_response::{IntraTransactions, QueryByAccount}; @@ -20,14 +20,17 @@ async fn handle_event( ) -> Result, StatusCode> { let client_request = event.0; - let conn = pool.get_conn().await; + let conn = pool + .get_conn() + .await + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; let svc = Service::new(&conn, None); let account = client_request.account_name; - let record_limit = env::var("RETURN_RECORD_LIMIT") - .unwrap_or_else(|_| panic!("RETURN_RECORD_LIMIT variable assignment")) + let record_limit = envvar::required("RETURN_RECORD_LIMIT") + .unwrap() .parse::() .unwrap(); @@ -52,8 +55,7 @@ async fn handle_event( async fn main() { tracing_subscriber::fmt::init(); - let readiness_check_path = env::var(READINESS_CHECK_PATH) - .unwrap_or_else(|_| panic!("{READINESS_CHECK_PATH} variable assignment")); + let readiness_check_path = envvar::required(READINESS_CHECK_PATH).unwrap(); let conn_uri = DB::create_conn_uri_from_env_vars(); @@ -67,9 +69,9 @@ async fn main() { ) .with_state(pool); - let hostname_or_ip = env::var("HOSTNAME_OR_IP").unwrap_or("0.0.0.0".to_string()); + let hostname_or_ip = envvar::optional("HOSTNAME_OR_IP", "0.0.0.0"); - let port = env::var("TRANSACTIONS_BY_ACCOUNT_PORT").unwrap(); + let port = envvar::required("TRANSACTIONS_BY_ACCOUNT_PORT").unwrap(); let serve_addr = format!("{hostname_or_ip}:{port}"); From d1a0f0c05928d1b30791944c7e9e42d7556a6bd6 Mon Sep 17 00:00:00 2001 From: max funk Date: Tue, 3 Mar 2026 08:01:19 -0800 Subject: [PATCH 09/12] envvar project config --- cue/project_conf.cue | 1 + project.yaml | 10 +++++++++- 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/cue/project_conf.cue b/cue/project_conf.cue index db28ab79c..439c4531a 100644 --- a/cue/project_conf.cue +++ b/cue/project_conf.cue @@ -64,6 +64,7 @@ client!: { crates: { #Dir cache: #Lib + envvar: #Lib httpclient: #Lib pg: #Lib pubsub: #Lib diff --git a/project.yaml b/project.yaml index b23b7d1c3..677475ae4 100644 --- a/project.yaml +++ b/project.yaml @@ -57,6 +57,14 @@ crates: set: {} get: [] params: [] + envvar: + runtime: rust1.x + min_code_cov: 0 + type: lib + env_var: + set: {} + get: [] + params: [] httpclient: runtime: rust1.x min_code_cov: 55 @@ -961,4 +969,4 @@ env_var: get: - CLIENT_URI - GRAPHQL_URI -params: [] +params: [] \ No newline at end of file From 0bb4286916424be871d5140f885006126da3d2c5 Mon Sep 17 00:00:00 2001 From: max funk Date: Tue, 3 Mar 2026 08:02:34 -0800 Subject: [PATCH 10/12] error integration tests --- tests/src/helpers.rs | 7 ++- tests/src/integration_tests.rs | 91 ++++++++++++++++++++++++++++++++++ tests/src/requests.rs | 54 ++++++++++++++------ 3 files changed, 132 insertions(+), 20 deletions(-) diff --git a/tests/src/helpers.rs b/tests/src/helpers.rs index 082ec96e4..886e84e93 100644 --- a/tests/src/helpers.rs +++ b/tests/src/helpers.rs @@ -55,11 +55,10 @@ pub async fn create_transaction() -> Transaction { ); let approve_request_body_json = json!(approve_request_body).to_string(); let client = Client::new(); - let approve_request_response = client + let approve_request_response_body = client .post(approve_request_uri, approve_request_body_json) .await .unwrap(); - let approve_request_response_body = approve_request_response.text().await.unwrap(); let approve_request: IntraTransaction = serde_json::from_str(&approve_request_response_body).unwrap(); approve_request.transaction @@ -72,7 +71,7 @@ pub async fn get_cached_account_rule(account: &str) -> String { if env::var("AWS_LAMBDA_FUNCTION_NAME").is_ok() { ddb_query(&key).await.remove(0) } else { - let client = cache::RedisClient::new().await; + let client = cache::RedisClient::new().await.unwrap(); client.init().await.unwrap(); client.smembers(&key).await.unwrap().remove(0) } @@ -85,7 +84,7 @@ pub async fn set_cached_account_rule(account: &str, old: &str, new: &str) { let sk = old_json["id"].as_str().unwrap(); ddb_put(&key, sk, new).await; } else { - let client = cache::RedisClient::new().await; + let client = cache::RedisClient::new().await.unwrap(); client.init().await.unwrap(); client.srem(&key, old).await.unwrap(); client.sadd(&key, new).await.unwrap(); diff --git a/tests/src/integration_tests.rs b/tests/src/integration_tests.rs index 7b73d2ea8..4e74a268b 100644 --- a/tests/src/integration_tests.rs +++ b/tests/src/integration_tests.rs @@ -618,6 +618,97 @@ mod tests { } } + #[tokio::test] + #[cfg_attr(not(feature = "integration_tests"), ignore)] + async fn it_returns_400_from_request_create_on_missing_author() { + _before_each(); + + let empty_items = types::transaction_item::TransactionItems(vec![ + types::transaction_item::TransactionItem { + id: None, + transaction_id: None, + item_id: "bread".to_string(), + price: "3.000".to_string(), + quantity: "2.000".to_string(), + rule_instance_id: None, + rule_exec_ids: None, + unit_of_measurement: None, + units_measured: None, + debitor: "JacobWebb".to_string(), + creditor: "GroceryStore".to_string(), + debitor_profile_id: None, + creditor_profile_id: None, + debitor_approval_time: None, + creditor_approval_time: None, + debitor_rejection_time: None, + creditor_rejection_time: None, + debitor_expiration_time: None, + creditor_expiration_time: None, + approvals: None, + }, + ]); + + let err = r::create_request_http_raw("GroceryStore".to_string(), empty_items) + .await + .unwrap_err(); + + match err { + httpclient::ClientError::Downstream { status, .. } => assert_eq!(status, 400), + _ => panic!("expected downstream error"), + } + } + + #[tokio::test] + #[cfg_attr(not(feature = "integration_tests"), ignore)] + async fn it_returns_400_from_request_approve_on_invalid_id() { + _before_each(); + + let err = r::approve_request_http_raw( + "not_a_number".to_string(), + "JacobWebb".to_string(), + "debitor".to_string(), + "JacobWebb".to_string(), + ) + .await + .unwrap_err(); + + match err { + httpclient::ClientError::Downstream { status, message } => { + assert_eq!(status, 400); + assert!(message.contains("invalid request id")); + } + _ => panic!("expected downstream error"), + } + } + + #[tokio::test] + #[cfg_attr(not(feature = "integration_tests"), ignore)] + async fn it_returns_400_from_request_approve_on_previously_approved() { + _before_each(); + + let transaction = h::create_transaction().await; + let transaction_id = transaction.id.unwrap(); + + let debitor = transaction.transaction_items.0[0].debitor.clone(); + + let err = r::approve_request_http_raw( + transaction_id, + debitor.clone(), + "debitor".to_string(), + debitor, + ) + .await + .unwrap_err(); + + match err { + httpclient::ClientError::Downstream { status, message } => { + assert_eq!(status, 400); + assert!(message.contains("previously approved")); + } + _ => panic!("expected downstream error"), + } + } + #[tokio::test] #[cfg_attr(not(feature = "integration_tests"), ignore)] async fn it_reads_rules_from_cache() { diff --git a/tests/src/requests.rs b/tests/src/requests.rs index 455b6ea0a..051355e22 100644 --- a/tests/src/requests.rs +++ b/tests/src/requests.rs @@ -604,12 +604,41 @@ pub async fn get_request_by_id_gql( res.request_by_id } +pub async fn create_request_http_raw( + auth_account: String, + transaction_items: TransactionItems, +) -> Result { + let transaction = Transaction::new(auth_account.clone(), None, transaction_items); + let body = IntraTransaction::new(auth_account.clone(), transaction); + let body_json = json!(body).to_string(); + let client = Client::new(); + let uri = Uri::new_from_env_var("REQUEST_CREATE_URL").to_string(); + client.post(uri, body_json).await +} + +pub async fn approve_request_http_raw( + id: String, + account_name: String, + account_role: String, + auth_account: String, +) -> Result { + let body_json = json!({ + "id": id, + "account_name": account_name, + "account_role": account_role, + "auth_account": auth_account + }) + .to_string(); + let client = Client::new(); + let uri = Uri::new_from_env_var("REQUEST_APPROVE_URL").to_string(); + client.post(uri, body_json).await +} + pub async fn get_rules_http(transaction: Transaction) -> IntraTransaction { let client = Client::new(); let uri = Uri::new_from_env_var("RULE_URL").to_string(); let body_json = json!(transaction).to_string(); - let response = client.post(uri, body_json).await.unwrap(); - let response_string = response.text().await.unwrap(); + let response_string = client.post(uri, body_json).await.unwrap(); serde_json::from_str(&response_string).unwrap() } @@ -622,8 +651,7 @@ pub async fn create_request_http( let body_json = json!(body).to_string(); let client = Client::new(); let uri = Uri::new_from_env_var("REQUEST_CREATE_URL").to_string(); - let response = client.post(uri, body_json).await.unwrap(); - let response_string = response.text().await.unwrap(); + let response_string = client.post(uri, body_json).await.unwrap(); let transaction_request: IntraTransaction = serde_json::from_str(&response_string).unwrap(); transaction_request.transaction } @@ -643,8 +671,7 @@ pub async fn approve_request_http( .to_string(); let client = Client::new(); let uri = Uri::new_from_env_var("REQUEST_APPROVE_URL").to_string(); - let response = client.post(uri, body_json).await.unwrap(); - let response_string = response.text().await.unwrap(); + let response_string = client.post(uri, body_json).await.unwrap(); let approved: IntraTransaction = serde_json::from_str(&response_string).unwrap(); approved.transaction } @@ -657,8 +684,7 @@ pub async fn get_account_balance_http(account_name: String, auth_account: String .to_string(); let client = Client::new(); let uri = Uri::new_from_env_var("BALANCE_BY_ACCOUNT_URL").to_string(); - let response = client.post(uri, body_json).await.unwrap(); - let response_string = response.text().await.unwrap(); + let response_string = client.post(uri, body_json).await.unwrap(); let balance_float: f32 = serde_json::from_str(&response_string).unwrap(); format!("{balance_float:.3}") } @@ -676,8 +702,7 @@ pub async fn get_transaction_by_id_http( .to_string(); let client = Client::new(); let uri = Uri::new_from_env_var("TRANSACTION_BY_ID_URL").to_string(); - let response = client.post(uri, body_json).await.unwrap(); - let response_string = response.text().await.unwrap(); + let response_string = client.post(uri, body_json).await.unwrap(); let intra_transaction: IntraTransaction = serde_json::from_str(&response_string).unwrap(); intra_transaction } @@ -693,8 +718,7 @@ pub async fn get_transactions_by_account_http( .to_string(); let client = Client::new(); let uri = Uri::new_from_env_var("TRANSACTIONS_BY_ACCOUNT_URL").to_string(); - let response = client.post(uri, body_json).await.unwrap(); - let response_string = response.text().await.unwrap(); + let response_string = client.post(uri, body_json).await.unwrap(); let intra_transactions: IntraTransactions = serde_json::from_str(&response_string).unwrap(); intra_transactions } @@ -712,8 +736,7 @@ pub async fn get_request_by_id_http( .to_string(); let client = Client::new(); let uri = Uri::new_from_env_var("REQUEST_BY_ID_URL").to_string(); - let response = client.post(uri, body_json).await.unwrap(); - let response_string = response.text().await.unwrap(); + let response_string = client.post(uri, body_json).await.unwrap(); let intra_transaction: IntraTransaction = serde_json::from_str(&response_string).unwrap(); intra_transaction } @@ -729,8 +752,7 @@ pub async fn get_requests_by_account_http( .to_string(); let client = Client::new(); let uri = Uri::new_from_env_var("REQUESTS_BY_ACCOUNT_URL").to_string(); - let response = client.post(uri, body_json).await.unwrap(); - let response_string = response.text().await.unwrap(); + let response_string = client.post(uri, body_json).await.unwrap(); let intra_transactions: IntraTransactions = serde_json::from_str(&response_string).unwrap(); intra_transactions } From 6110a2b644d264a86825fb4c53cd3c5b74bf22b3 Mon Sep 17 00:00:00 2001 From: max funk Date: Tue, 3 Mar 2026 08:07:54 -0800 Subject: [PATCH 11/12] summary build with a pipeline --- .../modules/codepipeline/v001/codepipeline.tf | 84 +++++++------------ scripts/ecr-images.sh | 50 ++++++----- 2 files changed, 56 insertions(+), 78 deletions(-) diff --git a/infra/terraform/aws/modules/codepipeline/v001/codepipeline.tf b/infra/terraform/aws/modules/codepipeline/v001/codepipeline.tf index e1906e72b..493edb2a9 100644 --- a/infra/terraform/aws/modules/codepipeline/v001/codepipeline.tf +++ b/infra/terraform/aws/modules/codepipeline/v001/codepipeline.tf @@ -18,6 +18,21 @@ resource "aws_codepipeline" "build" { type = "S3" } + variable { + name = "RUN_TESTS" + default_value = "true" + } + + variable { + name = "PUSH_IMAGE" + default_value = "true" + } + + variable { + name = "DEPLOY" + default_value = "false" + } + stage { name = "Source" action { @@ -50,31 +65,29 @@ resource "aws_codepipeline" "build" { run_order = 1 configuration = { ProjectName = action.value + EnvironmentVariables = jsonencode([ + { + name = "RUN_TESTS" + value = "#{variables.RUN_TESTS}" + type = "PLAINTEXT" + }, + { + name = "PUSH_IMAGE" + value = "#{variables.PUSH_IMAGE}" + type = "PLAINTEXT" + }, + { + name = "DEPLOY" + value = "#{variables.DEPLOY}" + type = "PLAINTEXT" + } + ]) } } } } } -resource "aws_cloudwatch_event_rule" "s3_trigger" { - name = "mxfactorial-build-trigger-${local.ID_ENV}" - - event_pattern = jsonencode({ - source = ["aws.s3"] - detail-type = ["Object Created"] - detail = { - bucket = { name = [var.artifacts_bucket_name] } - object = { key = [{ prefix = local.BUILD_SOURCE_LOCATION }] } - } - }) -} - -resource "aws_cloudwatch_event_target" "codepipeline" { - rule = aws_cloudwatch_event_rule.s3_trigger.name - arn = aws_codepipeline.build.arn - role_arn = aws_iam_role.eventbridge_pipeline.arn -} - resource "aws_iam_role" "codepipeline" { name = "codepipeline-build-${local.ID_ENV}" @@ -123,36 +136,3 @@ resource "aws_iam_role_policy" "codepipeline" { ] }) } - -resource "aws_iam_role" "eventbridge_pipeline" { - name = "eventbridge-pipeline-${local.ID_ENV}" - - assume_role_policy = jsonencode({ - Version = "2012-10-17" - Statement = [ - { - Action = "sts:AssumeRole" - Effect = "Allow" - Principal = { - Service = "events.amazonaws.com" - } - } - ] - }) -} - -resource "aws_iam_role_policy" "eventbridge_pipeline" { - name = "eventbridge-pipeline-${local.ID_ENV}" - role = aws_iam_role.eventbridge_pipeline.id - - policy = jsonencode({ - Version = "2012-10-17" - Statement = [ - { - Effect = "Allow" - Action = "codepipeline:StartPipelineExecution" - Resource = aws_codepipeline.build.arn - } - ] - }) -} diff --git a/scripts/ecr-images.sh b/scripts/ecr-images.sh index f1ffffa94..86509716f 100644 --- a/scripts/ecr-images.sh +++ b/scripts/ecr-images.sh @@ -104,37 +104,24 @@ function start_builds() { source scripts/zip-services.sh + echo "*** uploading archive to s3://$ARTIFACTS_BUCKET/$BUILD_OBJECT_KEY_PATH/" + aws s3 cp $SERVICES_ZIP s3://$ARTIFACTS_BUCKET/$BUILD_OBJECT_KEY_PATH/ --region $REGION + rm $SERVICES_ZIP + BUILD_IDS="" for SVC in $(get_services); do - UPLOAD_KEY_PATH="$BUILD_OBJECT_KEY_PATH/$SVC" - - echo "*** uploading archive to s3://$ARTIFACTS_BUCKET/$UPLOAD_KEY_PATH/" - aws s3 cp $SERVICES_ZIP s3://$ARTIFACTS_BUCKET/$UPLOAD_KEY_PATH/ --region $REGION - - DEPLOY_TARGET=$(yq ".. | select(has(\"$SVC\")) | .$SVC.deploy_target // \"\"" $PROJECT_CONF | head -1) - DEPLOY_LAMBDA=false - DEPLOY_ECS=false - if [[ "$DO_DEPLOY" == "true" ]]; then - if [[ "$DEPLOY_TARGET" == "lambda" ]]; then - DEPLOY_LAMBDA=true - elif [[ "$DEPLOY_TARGET" == "ecs" ]]; then - DEPLOY_ECS=true - fi - fi - PROJECT_NAME="mxfactorial-$SVC-$ID_ENV" echo -e "${YELLOW}starting $PROJECT_NAME...${RESET}" BUILD_ID=$(aws codebuild start-build \ --project-name $PROJECT_NAME \ --region $REGION \ --source-type-override S3 \ - --source-location-override "$ARTIFACTS_BUCKET/$UPLOAD_KEY_PATH/$SERVICES_ZIP" \ + --source-location-override "$ARTIFACTS_BUCKET/$BUILD_OBJECT_KEY_PATH/$SERVICES_ZIP" \ --artifacts-override type=NO_ARTIFACTS \ --environment-variables-override \ name=RUN_TESTS,value=$RUN_TESTS \ name=PUSH_IMAGE,value=$PUSH_IMAGE \ - name=DEPLOY_LAMBDA,value=$DEPLOY_LAMBDA \ - name=DEPLOY_ECS,value=$DEPLOY_ECS \ + name=DEPLOY,value=$DO_DEPLOY \ --query 'build.id' \ --output text) BUILD_IDS="$BUILD_IDS $BUILD_ID" @@ -144,8 +131,6 @@ function start_builds() { echo "" done - rm $SERVICES_ZIP - echo "*** waiting for builds to complete" echo -e "${YELLOW}(ctrl+c to exit - builds will continue in background)${RESET}" echo "" @@ -167,7 +152,11 @@ function start_builds() { done } -function trigger_pipeline() { +function start_pipeline() { + local RUN_TESTS=$1 + local PUSH_IMAGE=$2 + local DO_DEPLOY=$3 + ARTIFACTS_BUCKET_PREFIX=$(yq '.infra.terraform.aws.modules["project-storage"].env_var.set.ARTIFACTS_BUCKET_PREFIX.default' $PROJECT_CONF) ARTIFACTS_BUCKET="$ARTIFACTS_BUCKET_PREFIX-$ID_ENV" SERVICES_ZIP=$(yq '.scripts.env_var.set.SERVICES_ZIP.default' $PROJECT_CONF) @@ -179,7 +168,16 @@ function trigger_pipeline() { aws s3 cp $SERVICES_ZIP s3://$ARTIFACTS_BUCKET/build/ --region $REGION rm $SERVICES_ZIP - echo "*** codepipeline $PIPELINE_NAME triggered via eventbridge" + echo "*** starting codepipeline $PIPELINE_NAME" + aws codepipeline start-pipeline-execution \ + --name $PIPELINE_NAME \ + --region $REGION \ + --variables \ + name=RUN_TESTS,value=$RUN_TESTS \ + name=PUSH_IMAGE,value=$PUSH_IMAGE \ + name=DEPLOY,value=$DO_DEPLOY \ + --output text > /dev/null + echo "https://$REGION.console.aws.amazon.com/codesuite/codepipeline/pipelines/$PIPELINE_NAME/view/?region=$REGION" sleep 5 @@ -327,9 +325,9 @@ if [[ $BUILD == true ]]; then echo "" fi - # use pipeline for full builds (all services + push + no deploy), otherwise direct codebuild - if [[ -z "$SERVICE" && $PUSH == true && $TEST == true && $DEPLOY == false ]]; then - trigger_pipeline + # use pipeline for all-service builds, direct codebuild for single service + if [[ -z "$SERVICE" ]]; then + start_pipeline "$RUN_TESTS" "$PUSH_IMAGE" "$DO_DEPLOY" else start_builds "$RUN_TESTS" "$PUSH_IMAGE" "$DO_DEPLOY" fi From aa97c760049808ce99a771494957d470887d2099 Mon Sep 17 00:00:00 2001 From: max funk Date: Tue, 3 Mar 2026 08:08:56 -0800 Subject: [PATCH 12/12] support deploy_target config in codebuild --- .agents/onboard.md | 4 ++-- infra/terraform/aws/modules/ci/v001/main.tf | 8 ++++++++ infra/terraform/aws/modules/codebuild/v001/codebuild.tf | 6 +++--- infra/terraform/aws/modules/codebuild/v001/variables.tf | 3 +++ .../terraform/aws/modules/ecr/v001/buildspecs/build.yaml | 5 +++++ 5 files changed, 21 insertions(+), 5 deletions(-) diff --git a/.agents/onboard.md b/.agents/onboard.md index 5759cf200..fb23db109 100644 --- a/.agents/onboard.md +++ b/.agents/onboard.md @@ -138,8 +138,8 @@ use yq instead of jq for local json/yaml scripting codebuild projects are in infra/terraform/aws/modules/ci/v001. per-service builds in main.tf (sources codebuild module), integ tests in integ.tf (buildspec inlined). ci module is only used by init-dev -buildspecs use runtime config via env vars (RUN_TESTS, PUSH_IMAGE, DEPLOY_LAMBDA, DEPLOY_ECS) set by `--environment-variables-override` in ecr-images.sh +buildspecs use runtime config via env vars (RUN_TESTS, PUSH_IMAGE, DEPLOY, DEPLOY_TARGET). DEPLOY_TARGET is set per codebuild project in terraform. DEPLOY_LAMBDA/DEPLOY_ECS are derived in the buildspec from DEPLOY + DEPLOY_TARGET -s3 upload to artifacts bucket auto-triggers codepipeline via eventbridge when using `--build --push` without `--service` +`--build` without `--service` triggers codepipeline via `start-pipeline-execution` with pipeline variables. `--build --service ` starts a direct codebuild job with `--environment-variables-override` say "design" instead of "approach" \ No newline at end of file diff --git a/infra/terraform/aws/modules/ci/v001/main.tf b/infra/terraform/aws/modules/ci/v001/main.tf index 899a576cd..18b1f2fa1 100644 --- a/infra/terraform/aws/modules/ci/v001/main.tf +++ b/infra/terraform/aws/modules/ci/v001/main.tf @@ -7,6 +7,13 @@ locals { SERVICES_ZIP = local.PROJECT_CONF.scripts.env_var.set.SERVICES_ZIP.default INTEG_TEST_OBJECT_KEY_PATH = local.STORAGE_ENV_VAR.INTEG_TEST_OBJECT_KEY_PATH.default INTEG_SOURCE_LOCATION = "${local.INTEG_TEST_OBJECT_KEY_PATH}/${local.SERVICES_ZIP}" + + # map service name -> deploy_target from project.yaml + service_deploy_targets = merge( + { for k, v in try(local.PROJECT_CONF.services, {}) : k => v.deploy_target if try(v.type, "") == "app" }, + { for k, v in try(local.PROJECT_CONF.migrations, {}) : k => v.deploy_target if try(v.type, "") == "app" }, + try(local.PROJECT_CONF.client.type, "") == "app" ? { "client" = local.PROJECT_CONF.client.deploy_target } : {} + ) } data "aws_caller_identity" "current" {} @@ -27,6 +34,7 @@ module "codebuild_projects" { artifacts_bucket_arn = var.artifacts_bucket_arn ecr_repository_arn = "arn:aws:ecr:${data.aws_region.current.id}:${data.aws_caller_identity.current.account_id}:repository/${local.ID_ENV_PREFIX}/${each.value}" lambda_function_arn = "arn:aws:lambda:${data.aws_region.current.id}:${data.aws_caller_identity.current.account_id}:function:${each.value}-${local.ID_ENV}" + deploy_target = try(local.service_deploy_targets[each.value], "none") } module "codepipeline" { diff --git a/infra/terraform/aws/modules/codebuild/v001/codebuild.tf b/infra/terraform/aws/modules/codebuild/v001/codebuild.tf index c992b62b5..81b5c9dbf 100644 --- a/infra/terraform/aws/modules/codebuild/v001/codebuild.tf +++ b/infra/terraform/aws/modules/codebuild/v001/codebuild.tf @@ -41,12 +41,12 @@ resource "aws_codebuild_project" "default" { value = "true" } environment_variable { - name = "DEPLOY_LAMBDA" + name = "DEPLOY" value = "false" } environment_variable { - name = "DEPLOY_ECS" - value = "false" + name = "DEPLOY_TARGET" + value = var.deploy_target } } diff --git a/infra/terraform/aws/modules/codebuild/v001/variables.tf b/infra/terraform/aws/modules/codebuild/v001/variables.tf index e930c5a48..449b3649c 100644 --- a/infra/terraform/aws/modules/codebuild/v001/variables.tf +++ b/infra/terraform/aws/modules/codebuild/v001/variables.tf @@ -9,6 +9,9 @@ variable "buildspec" {} variable "artifacts_bucket_arn" {} variable "ecr_repository_arn" {} variable "lambda_function_arn" {} +variable "deploy_target" { + default = "none" +} variable "ecs_cluster_arn" { default = "" } diff --git a/infra/terraform/aws/modules/ecr/v001/buildspecs/build.yaml b/infra/terraform/aws/modules/ecr/v001/buildspecs/build.yaml index 690132325..a251b40cd 100644 --- a/infra/terraform/aws/modules/ecr/v001/buildspecs/build.yaml +++ b/infra/terraform/aws/modules/ecr/v001/buildspecs/build.yaml @@ -6,6 +6,11 @@ phases: - aws ecr get-login-password --region $REGION | docker login --username AWS --password-stdin $AWS_ACCOUNT_ID.dkr.ecr.$REGION.amazonaws.com - export SHORT_SHA=$(echo ${CODEBUILD_RESOLVED_SOURCE_VERSION:-$CODEBUILD_BUILD_ID} | sed 's/.*://' | cut -c1-7) - export ECR_URI=$AWS_ACCOUNT_ID.dkr.ecr.$REGION.amazonaws.com/$ENV_ID/$ENV/$SERVICE_NAME + - | + if [ "${DEPLOY:-false}" = "true" ]; then + if [ "$DEPLOY_TARGET" = "lambda" ]; then export DEPLOY_LAMBDA=true; fi + if [ "$DEPLOY_TARGET" = "ecs" ]; then export DEPLOY_ECS=true; fi + fi build: commands: - echo "building $SERVICE_NAME (RUN_TESTS=$RUN_TESTS, PUSH_IMAGE=$PUSH_IMAGE)..."