feat(rds): support PostgreSQL aws_lambda extension (#800)

README.md

@@ -69,7 +69,7 @@ Other install options (Cargo, Docker, Docker Compose, source) are documented at
 | SES (v2 + v1 inbound)  | 110 | Sending, templates, DKIM, **real receipt rule execution**              |
 | Cognito User Pools     | 122 | Pools, clients, MFA, identity providers, full auth flows; verification email -> SES, SMS -> SNS, all 12 Lambda triggers |
 | Kinesis                |  39 | Streams, records, shard iterators, retention                           |
-| RDS                    | 163 | Real Postgres, MySQL, MariaDB, Oracle, SQL Server, Db2 via Docker; lifecycle ops emit `aws.rds` EventBridge events |
+| RDS                    | 163 | Real Postgres, MySQL, MariaDB, Oracle, SQL Server, Db2 via Docker; lifecycle ops emit `aws.rds` EventBridge events; PostgreSQL `aws_lambda` extension invokes fakecloud Lambda functions from SQL |
 | ElastiCache            |  75 | Real Redis, Valkey, Memcached via Docker                               |
 | Step Functions         |  37 | Full ASL interpreter, Lambda/SQS/SNS/EventBridge/DynamoDB tasks        |
 | API Gateway v1         | 124 | REST APIs, resources, methods, integrations (`MOCK`/`HTTP`/`HTTP_PROXY`/`AWS_PROXY` Lambda), deployments, stages, API keys, usage plans, authorizers, models, request validators, VPC links, domain names, base path mappings, client certs, gateway responses, docs, tags |
@@ -119,7 +119,7 @@ Full guides: [fakecloud.dev/docs/guides](https://fakecloud.dev/docs/guides).
 | Cognito User Pools  | 122 operations                                     | [Paid only](https://docs.localstack.cloud/references/licensing/)               |
 | SES v2              | Full send + templates + DKIM + suppression         | [Paid only](https://docs.localstack.cloud/references/licensing/)               |
 | SES inbound email   | Real receipt rule action execution                 | [Stored but never executed](https://docs.localstack.cloud/user-guide/aws/ses/) |
-| RDS                 | 163 operations, PostgreSQL/MySQL/MariaDB/Oracle/SQL Server/Db2 via Docker | [Paid only](https://docs.localstack.cloud/references/licensing/)               |
+| RDS                 | 163 operations, PostgreSQL/MySQL/MariaDB/Oracle/SQL Server/Db2 via Docker, PostgreSQL `aws_lambda` extension | [Paid only](https://docs.localstack.cloud/references/licensing/)               |
 | ElastiCache         | 75 operations, Redis, Valkey, and Memcached via Docker | [Paid only](https://docs.localstack.cloud/references/licensing/)           |
 | API Gateway v1      | 124 operations — REST APIs incl. real Lambda proxy data plane | [Paid only](https://docs.localstack.cloud/references/licensing/)               |
 | API Gateway v2      | 103 operations — HTTP APIs + developer portals     | [Paid only](https://docs.localstack.cloud/references/licensing/)               |

crates/fakecloud-e2e/tests/rds_aws_lambda.rs

@@ -0,0 +1,174 @@
+//! End-to-end tests for the RDS PostgreSQL `aws_lambda` extension.
+//!
+//! Drives a full happy path: create a Lambda, create a Postgres DB
+//! instance (which triggers the lazy build of `fakecloud-postgres`),
+//! connect via tokio_postgres, run `CREATE EXTENSION aws_lambda CASCADE`,
+//! and call `aws_lambda.invoke()` with both a name and an
+//! `aws_commons.create_lambda_function_arn` composite. Async (`Event`)
+//! invocation path is exercised too.
+
+mod helpers;
+
+use std::io::Write;
+
+use aws_sdk_lambda::primitives::Blob;
+use helpers::TestServer;
+use tokio_postgres::NoTls;
+
+fn make_echo_zip() -> Vec<u8> {
+    // Returns the raw event back to the caller so we can verify the
+    // payload round-trips through plpython3u + the bridge endpoint.
+    let buf = Vec::new();
+    let cursor = std::io::Cursor::new(buf);
+    let mut writer = zip::ZipWriter::new(cursor);
+    let options = zip::write::SimpleFileOptions::default();
+    writer.start_file("index.py", options).unwrap();
+    writer
+        .write_all(b"def handler(event, context):\n    return event\n")
+        .unwrap();
+    let cursor = writer.finish().unwrap();
+    cursor.into_inner()
+}
+
+async fn connect_with_retry(
+    host: &str,
+    port: i32,
+    user: &str,
+    password: &str,
+    dbname: &str,
+) -> tokio_postgres::Client {
+    let connection_string =
+        format!("host={host} port={port} user={user} password={password} dbname={dbname}");
+    let mut last_error = None;
+    for _ in 0..30 {
+        match tokio_postgres::connect(&connection_string, NoTls).await {
+            Ok((client, connection)) => {
+                tokio::spawn(async move {
+                    let _ = connection.await;
+                });
+                return client;
+            }
+            Err(error) => {
+                last_error = Some(error);
+                tokio::time::sleep(std::time::Duration::from_millis(500)).await;
+            }
+        }
+    }
+    panic!(
+        "could not connect to postgres at {}:{}: {:?}",
+        host, port, last_error
+    );
+}
+
+#[tokio::test]
+async fn aws_lambda_extension_invoke_round_trip() {
+    let server = TestServer::start().await;
+    let lambda = server.lambda_client().await;
+    let rds = server.rds_client().await;
+
+    // 1. Create the echo Lambda.
+    lambda
+        .create_function()
+        .function_name("echo")
+        .runtime(aws_sdk_lambda::types::Runtime::Python312)
+        .role("arn:aws:iam::000000000000:role/test-role")
+        .handler("index.handler")
+        .code(
+            aws_sdk_lambda::types::FunctionCode::builder()
+                .zip_file(Blob::new(make_echo_zip()))
+                .build(),
+        )
+        .send()
+        .await
+        .expect("create echo lambda");
+
+    // 2. Create the Postgres DB instance — triggers lazy fakecloud-postgres
+    //    image build on first run, so this can take a while.
+    let create = rds
+        .create_db_instance()
+        .db_instance_identifier("aws-lambda-ext-db")
+        .allocated_storage(20)
+        .db_instance_class("db.t3.micro")
+        .engine("postgres")
+        .engine_version("16.3")
+        .master_username("admin")
+        .master_user_password("secret123")
+        .db_name("appdb")
+        .send()
+        .await
+        .expect("create postgres instance");
+
+    let endpoint = create
+        .db_instance()
+        .and_then(|i| i.endpoint())
+        .expect("endpoint");
+    let host = endpoint.address().expect("address").to_string();
+    let port = endpoint.port().expect("port");
+
+    // 3. Connect and load the extension.
+    let client = connect_with_retry(&host, port, "admin", "secret123", "appdb").await;
+    client
+        .simple_query("CREATE EXTENSION IF NOT EXISTS aws_lambda CASCADE")
+        .await
+        .expect("load aws_lambda extension");
+
+    // tokio-postgres in this workspace doesn't ship the with-serde_json-1
+    // feature, so we can't bind a Rust `&str` to a postgres `json`
+    // parameter. Embed payloads as SQL literals (test fixtures, no
+    // injection concern) and cast result `payload` to text on the wire.
+
+    // 4. Sync invoke by function name + json payload — payload round-trips.
+    let row = client
+        .query_one(
+            "SELECT status_code, payload::text \
+             FROM aws_lambda.invoke('echo', '{\"hello\":\"world\"}'::json)",
+            &[],
+        )
+        .await
+        .expect("invoke by name");
+    let status_code: i32 = row.get(0);
+    let payload_text: String = row.get(1);
+    let payload: serde_json::Value = serde_json::from_str(&payload_text).unwrap();
+    assert_eq!(status_code, 200);
+    assert_eq!(payload, serde_json::json!({"hello": "world"}));
+
+    // 5. aws_commons.create_lambda_function_arn returns a composite type.
+    let arn_row = client
+        .query_one(
+            "SELECT (aws_commons.create_lambda_function_arn('echo')).function_name",
+            &[],
+        )
+        .await
+        .expect("create_lambda_function_arn");
+    let function_name: String = arn_row.get(0);
+    assert_eq!(function_name, "echo");
+
+    // 6. Sync invoke via the composite-typed overload.
+    let row = client
+        .query_one(
+            "SELECT status_code, payload::text FROM aws_lambda.invoke(\
+                aws_commons.create_lambda_function_arn('echo'), '{\"k\":1}'::json)",
+            &[],
+        )
+        .await
+        .expect("invoke via composite arn");
+    let status_code: i32 = row.get(0);
+    let payload_text: String = row.get(1);
+    let payload: serde_json::Value = serde_json::from_str(&payload_text).unwrap();
+    assert_eq!(status_code, 200);
+    assert_eq!(payload, serde_json::json!({"k": 1}));
+
+    // 7. Async (Event) invocation returns 202 immediately.
+    let row = client
+        .query_one(
+            "SELECT status_code, payload::text FROM aws_lambda.invoke(\
+                'echo', '{\"async\":true}'::json, NULL, 'Event')",
+            &[],
+        )
+        .await
+        .expect("invoke async");
+    let status_code: i32 = row.get(0);
+    let payload_text: Option<String> = row.get(1);
+    assert_eq!(status_code, 202);
+    assert!(payload_text.is_none());
+}

crates/fakecloud-rds/Cargo.toml

@@ -25,3 +25,5 @@ tokio-postgres = { workspace = true }
 mysql_async = "0.34"
 base64 = { workspace = true }
 tracing = { workspace = true }
+sha2 = { workspace = true }
+tempfile = { workspace = true }

crates/fakecloud-rds/assets/postgres/Dockerfile

@@ -0,0 +1,13 @@
+ARG PG_VERSION=16
+FROM postgres:${PG_VERSION}
+ARG PG_VERSION
+
+RUN apt-get update \
+    && apt-get install -y --no-install-recommends \
+        postgresql-plpython3-${PG_VERSION} \
+        ca-certificates \
+    && rm -rf /var/lib/apt/lists/*
+
+COPY aws_commons.control aws_commons--1.0.sql \
+     aws_lambda.control aws_lambda--1.0.sql \
+     /usr/share/postgresql/${PG_VERSION}/extension/

crates/fakecloud-rds/assets/postgres/aws_commons--1.0.sql

@@ -0,0 +1,24 @@
+-- aws_commons extension v1.0 (fakecloud)
+-- Provides composite types and helpers used by aws_lambda and aws_s3 RDS extensions.
+
+\echo Use "CREATE EXTENSION aws_commons" to load this file. \quit
+
+CREATE TYPE aws_commons._lambda_function_arn_1 AS (
+    function_name text,
+    region text
+);
+
+CREATE FUNCTION aws_commons.create_lambda_function_arn(
+    function_name text,
+    region text DEFAULT NULL
+) RETURNS aws_commons._lambda_function_arn_1
+LANGUAGE plpgsql IMMUTABLE
+AS $$
+DECLARE
+    result aws_commons._lambda_function_arn_1;
+BEGIN
+    result.function_name := function_name;
+    result.region := region;
+    RETURN result;
+END;
+$$;

crates/fakecloud-rds/assets/postgres/aws_commons.control

@@ -0,0 +1,4 @@
+comment = 'AWS commons types and helpers (fakecloud)'
+default_version = '1.0'
+relocatable = false
+schema = 'aws_commons'

crates/fakecloud-rds/assets/postgres/aws_lambda--1.0.sql

@@ -0,0 +1,101 @@
+-- aws_lambda extension v1.0 (fakecloud)
+-- Calls fakecloud Lambda invocations through a host bridge endpoint.
+
+\echo Use "CREATE EXTENSION aws_lambda CASCADE" to load this file. \quit
+
+CREATE FUNCTION aws_lambda.invoke(
+    function_name text,
+    payload json,
+    region text DEFAULT NULL,
+    invocation_type text DEFAULT 'RequestResponse'
+) RETURNS TABLE(
+    status_code integer,
+    payload json,
+    executed_version text,
+    log_result text
+)
+LANGUAGE plpython3u
+AS $$
+import json
+import os
+import urllib.request
+import urllib.error
+
+endpoint = os.environ.get('FAKECLOUD_ENDPOINT')
+if not endpoint:
+    plpy.error('aws_lambda: FAKECLOUD_ENDPOINT not set on the database container')
+
+account_id = os.environ.get('FAKECLOUD_ACCOUNT_ID', '000000000000')
+default_region = os.environ.get('FAKECLOUD_REGION', 'us-east-1')
+
+body = {
+    'function_name': function_name,
+    'payload': json.loads(payload) if payload is not None else None,
+    'invocation_type': invocation_type,
+    'region': region or default_region,
+}
+
+req = urllib.request.Request(
+    endpoint.rstrip('/') + '/_fakecloud/rds/lambda-invoke',
+    data=json.dumps(body).encode('utf-8'),
+    headers={
+        'Content-Type': 'application/json',
+        'X-Fakecloud-Account-Id': account_id,
+    },
+    method='POST',
+)
+
+http_status = None
+try:
+    with urllib.request.urlopen(req, timeout=300) as resp:
+        raw = resp.read()
+        http_status = resp.status
+except urllib.error.HTTPError as e:
+    raw = e.read()
+    http_status = e.code
+except Exception as e:
+    plpy.error('aws_lambda: bridge call failed: {}'.format(e))
+
+try:
+    parsed = json.loads(raw)
+except ValueError:
+    parsed = {
+        'status_code': http_status,
+        'payload': {'errorMessage': raw.decode('utf-8', errors='replace')},
+    }
+
+status_code = parsed.get('status_code')
+if status_code is None:
+    status_code = http_status if http_status is not None else 0
+
+return [(
+    int(status_code),
+    json.dumps(parsed.get('payload')) if parsed.get('payload') is not None else None,
+    parsed.get('executed_version'),
+    parsed.get('log_result'),
+)]
+$$;
+
+-- LANGUAGE SQL keeps the user-facing arg name `payload` even though it
+-- also names a RETURNS TABLE column. PL/pgSQL would reject that as a
+-- duplicate identifier in the function namespace; SQL doesn't.
+CREATE FUNCTION aws_lambda.invoke(
+    function_name aws_commons._lambda_function_arn_1,
+    payload json,
+    region text DEFAULT NULL,
+    invocation_type text DEFAULT 'RequestResponse'
+) RETURNS TABLE(
+    status_code integer,
+    payload json,
+    executed_version text,
+    log_result text
+)
+LANGUAGE SQL
+AS $$
+    SELECT * FROM aws_lambda.invoke(
+        (function_name).function_name,
+        payload,
+        COALESCE(region, (function_name).region),
+        invocation_type
+    );
+$$;

crates/fakecloud-rds/assets/postgres/aws_lambda.control

@@ -0,0 +1,5 @@
+comment = 'AWS Lambda invocation from PostgreSQL (fakecloud)'
+default_version = '1.0'
+relocatable = false
+schema = 'aws_lambda'
+requires = 'plpython3u, aws_commons'