Add finish_payment logic

Cargo.toml

@@ -19,7 +19,7 @@ path = "src/finish_payment.rs"
 
 [dependencies]
 lambda_http = "0.8.1"
-tokio = { version = "1", features = ["macros"] }
+tokio = { version = "1", features = ["macros", "sync", "parking_lot"] }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 async-stripe = { version = "0.22", features = ["runtime-tokio-hyper"] }

infrastructure/lambda.tf

@@ -35,7 +35,6 @@ resource "aws_lambda_function" "finish_payment_lambda" {
     variables = {
       PAYMENTS_TABLE_NAME   = aws_dynamodb_table.payments.name
       STRIPE_SECRET_KEY     = var.STRIPE_API_KEY
-      STRIPE_WEBHOOK_SECRET = ""
     }
   }
 }

src/database/client.rs

@@ -0,0 +1,13 @@
+use aws_sdk_dynamodb::Client;
+use tokio::sync::OnceCell;
+
+static CLIENT: OnceCell<Client> = OnceCell::const_new();
+
+pub async fn get_client() -> &'static Client {
+    CLIENT
+        .get_or_init(|| async {
+            let config = aws_config::load_from_env().await;
+            Client::new(&config)
+        })
+        .await
+}

src/database/mod.rs

@@ -0,0 +1,5 @@
+pub mod client;
+pub mod payments_repository;
+
+pub use client::*;
+pub use payments_repository::*;

src/database/payments_repository.rs

@@ -0,0 +1,78 @@
+use aws_sdk_dynamodb::{types::AttributeValue, Client};
+use tokio::sync::OnceCell;
+
+use crate::{
+    domain::{Payment, PaymentStatus},
+    environment::{get_env_var, PAYMENTS_TABLE},
+};
+
+static PAYMENTS_REPOSITORY: OnceCell<PaymentsRepository> = OnceCell::const_new();
+
+#[derive(Clone)]
+pub struct PaymentsRepository {
+    client: Client,
+    table_name: String,
+}
+
+impl PaymentsRepository {
+    pub async fn get() -> PaymentsRepository {
+        PAYMENTS_REPOSITORY
+            .get_or_init(|| async {
+                let client = Client::new(&aws_config::load_from_env().await);
+                PaymentsRepository::new(client)
+            })
+            .await
+            .clone()
+    }
+
+    fn new(client: Client) -> Self {
+        let table_name = get_env_var(PAYMENTS_TABLE)
+            .unwrap_or_else(|_| format!("{} variable not set", PAYMENTS_TABLE));
+
+        Self { client, table_name }
+    }
+
+    pub async fn insert_payment(self, payment: Payment) -> Result<(), String> {
+        let id = AttributeValue::S(payment.id);
+        let amount = AttributeValue::N(payment.amount.to_string());
+        let sender = AttributeValue::S(payment.sender);
+        let status = AttributeValue::N((payment.status as i8).to_string());
+
+        let _request = self
+            .client
+            .put_item()
+            .table_name(self.table_name)
+            .item("id", id)
+            .item("amount", amount)
+            .item("sender", sender)
+            .item("status", status)
+            .send()
+            .await
+            .map_err(|e| e.to_string())?;
+
+        Ok(())
+    }
+
+    pub async fn update_payment_status(
+        self,
+        payment_id: &str,
+        new_status: PaymentStatus,
+    ) -> Result<(), String> {
+        let id = AttributeValue::S(String::from(payment_id));
+        let status = AttributeValue::N((new_status as i8).to_string());
+
+        let _request = self
+            .client
+            .update_item()
+            .table_name(self.table_name)
+            .key("id", id)
+            .update_expression("SET #status = :status")
+            .expression_attribute_names("#status", "status")
+            .expression_attribute_values(":status", status)
+            .send()
+            .await
+            .map_err(|e| e.to_string())?;
+
+        Ok(())
+    }
+}

src/finish_payment.rs

@@ -1,11 +1,63 @@
 use lambda_http::{service_fn, Body, Error, Request, Response};
+use serverless_payments::{
+    database::PaymentsRepository,
+    domain::PaymentStatus,
+    environment::{get_env_var, STRIPE_SECRET_KEY},
+    request_utils::get_header,
+};
+use stripe::{EventObject, Webhook};
+use tracing_subscriber::FmtSubscriber;
+
+const SIGNATURE_HEADER_KEY: &str = "Stripe-Signature";
 
 #[tokio::main]
 async fn main() -> Result<(), Error> {
+    FmtSubscriber::builder()
+        .with_max_level(tracing::Level::INFO)
+        .with_ansi(false)
+        .without_time()
+        .with_target(false)
+        .init();
     lambda_http::run(service_fn(handler)).await?;
     Ok(())
 }
 
-async fn handler(_event: Request) -> Result<Response<Body>, Error> {
-    Ok(Response::new(Body::from("Hello world!")))
+async fn handler(event: Request) -> Result<Response<Body>, Error> {
+    let signature = get_header(&event, SIGNATURE_HEADER_KEY)?;
+    let secret_key = get_env_var(STRIPE_SECRET_KEY)?;
+    let event_body = match event.body() {
+        Body::Text(s) => s,
+        _ => {
+            tracing::error!("Error getting event body");
+            return Err(Error::from("Error getting event body"));
+        }
+    };
+
+    let webhook_event = match Webhook::construct_event(event_body, &signature, &secret_key) {
+        Ok(webhook_event) => webhook_event,
+        Err(e) => {
+            tracing::error!("Error constructing webhook event: {:?}", e);
+            return Err(Error::from("Error constructing webhook event"));
+        }
+    };
+
+    let payment_intent_id = {
+        let payment_intent = match webhook_event.data.object {
+            // safe to unwrap, as this charge is the result of a PaymentIntent confirmation
+            EventObject::Charge(charge) => charge.payment_intent.unwrap(),
+            _ => {
+                tracing::error!("Error getting payment intent");
+                return Err(Error::from("Error getting payment intent"));
+            }
+        };
+
+        payment_intent.id()
+    };
+
+    let payment_repository = PaymentsRepository::get().await;
+    payment_repository
+        .update_payment_status(payment_intent_id.as_str(), PaymentStatus::Completed)
+        .await?;
+
+    Ok(Response::new(Body::from(())))
 }

src/initiate_payment.rs

@@ -1,7 +1,6 @@
-use aws_sdk_dynamodb::Client;
 use lambda_http::{service_fn, Body, Error, Request, Response};
 use serverless_payments::{
-    domain::PaymentRequest, payment_client::PaymentClient, payments_repository::PaymentsRepository,
+    database::PaymentsRepository, domain::PaymentRequest, payment_client::PaymentClient,
     request_utils::get_body,
 };
 use tracing_subscriber::FmtSubscriber;
@@ -28,9 +27,7 @@ async fn handler(event: Request) -> Result<Response<Body>, Error> {
     let redirect_url = payment_client.initiate_payment(&payment_request).await?;
 
     // Save the data to the database
-    let config = aws_config::load_from_env().await;
-    let db_client = Client::new(&config);
-    let payments_repository = PaymentsRepository::new(db_client);
+    let payments_repository = PaymentsRepository::get().await;
     payments_repository
         .insert_payment(payment_request.into())
         .await?;

src/lib.rs

@@ -1,5 +1,5 @@
+pub mod database;
 pub mod domain;
 pub mod environment;
 pub mod payment_client;
-pub mod payments_repository;
 pub mod request_utils;

src/request_utils.rs

@@ -14,3 +14,15 @@ where
     let result: T = serde_json::from_str(body).map_err(|e| e.to_string())?;
     Ok(result)
 }
+
+#[tracing::instrument]
+pub fn get_header(event: &Request, header: &str) -> Result<String, String> {
+    let header = event
+        .headers()
+        .get(header)
+        .ok_or_else(|| format!("Missing header: {}", header))?
+        .to_str()
+        .map_err(|e| e.to_string())?
+        .to_string();
+    Ok(header)
+}