use anyhow::Result;
use super::{Capability, CapabilityEdit, add_named_imports};
pub(crate) static EMAIL: Capability = Capability {
id: "email",
resource_type: "email",
summary: "send email through the platform (Amazon SES), with delivery, bounce and complaint events",
registration_key: "EmailClient",
receives_events: true,
add,
remove,
};
const HTTP_SOURCE: &str = "@forklaunch/core/http";
const HTTP_IMPORTS: &[&str] = &["createEmailClient", "type EmailClient"];
const RUNTIME_BLOCK: &str = "const configInjector = createConfigInjector(SchemaValidator(), {
EmailClient: {
lifetime: Lifetime.Scoped,
type: type<EmailClient>(),
factory: ({ PLATFORM_GATEWAY_URL, INSTANCE_ID, INSTANCE_HMAC_KEY }) =>
createEmailClient({
gatewayUrl: PLATFORM_GATEWAY_URL,
instanceId: INSTANCE_ID,
hmacKey: INSTANCE_HMAC_KEY
})
}
});";
pub(crate) const EVENTS_HANDLER: &str = "import type { EmailEventData, PlatformEvent } from '@forklaunch/core/http';
/**
* Email delivery events from the ForkLaunch platform (Amazon SES), verified
* by the platform-events controller before they reach this handler.
*
* email.delivered the recipient's mail server accepted the message
* email.bounced it was refused; `permanent` bounces are suppressed by
* the platform (a later send to the address answers 422)
* email.complained the recipient marked it as spam; also suppressed
*
* `data.messageId` is what EmailClient.send returned. Deliveries can repeat,
* so dedupe on `event.id`.
*/
const seen = new Set<string>();
export async function handle(event: PlatformEvent): Promise<void> {
// TODO: dedupe in the database when the service runs more than one replica.
if (seen.has(event.id)) return;
seen.add(event.id);
if (seen.size > 10_000) seen.delete(seen.values().next().value as string);
const data = event.data as unknown as EmailEventData;
switch (event.type) {
case 'email.delivered':
break;
case 'email.bounced':
case 'email.complained':
// TODO: mark data.recipients undeliverable (for example on the user
// record) so the app stops sending to them and can ask for a new address.
void data.recipients;
break;
default:
break;
}
}
";
fn add(edit: &mut CapabilityEdit) -> Result<()> {
edit.inject_registration("", HTTP_SOURCE, None, Some(RUNTIME_BLOCK))?;
let path = edit.registrations_path();
if let Some(before) = edit.read(&path)? {
let core_source = format!("@{}/core", edit.app_name);
let text = add_named_imports(&before, HTTP_SOURCE, HTTP_IMPORTS);
let text = add_named_imports(&text, &core_source, &["type"]);
if text != before {
edit.write(path, text);
}
}
edit.ensure_platform_events("email", EVENTS_HANDLER)?;
Ok(())
}
fn remove(edit: &mut CapabilityEdit) -> Result<()> {
edit.remove_registration(None, &[], &["EmailClient"])
}
#[cfg(test)]
mod tests {
use super::*;
use std::fs;
const REGISTRATIONS: &str = r#"import { SchemaValidator, number, string } from "@demo/core";
import { OpenTelemetryCollector } from "@forklaunch/core/http";
import { Lifetime, createConfigInjector, getEnvVar } from "@forklaunch/core/services";
const configInjector = createConfigInjector(SchemaValidator(), {
SERVICE_METADATA: {
lifetime: Lifetime.Singleton,
type: { name: string, version: string },
value: { name: "billing", version: "0.1.0" }
}
});
const environmentConfig = configInjector.chain({
PORT: {
lifetime: Lifetime.Singleton,
type: number,
value: Number(getEnvVar("PORT"))
}
});
const runtimeDependencies = environmentConfig.chain({
OpenTelemetryCollector: {
lifetime: Lifetime.Singleton,
type: OpenTelemetryCollector,
factory: () => new OpenTelemetryCollector("billing")
}
});
const serviceDependencies = runtimeDependencies.chain({});
"#;
const SERVER: &str = "import { forklaunchExpress, schemaValidator } from '@demo/core';\nimport { ci, tokens } from './bootstrapper';\nconst app = forklaunchExpress(schemaValidator, otel);\napp.use(billingRouter);\napp.listen(1);";
fn service(name: &str) -> (std::path::PathBuf, std::path::PathBuf) {
let root = std::env::temp_dir().join(format!("fl-cap-email-{name}"));
let _ = fs::remove_dir_all(&root);
let service = root.join("src/modules/billing");
fs::create_dir_all(&service).unwrap();
fs::write(service.join("registrations.ts"), REGISTRATIONS).unwrap();
fs::write(service.join("server.ts"), SERVER).unwrap();
(root, service)
}
fn edit(root: &std::path::Path, service: &std::path::Path, caps: &[&str]) -> CapabilityEdit {
let caps: Vec<String> = caps.iter().map(|c| c.to_string()).collect();
CapabilityEdit::new(root, "demo", "billing", service, caps.clone(), caps)
}
#[test]
fn add_registers_a_keyless_client_and_an_events_handler() {
let (root, service) = service("add");
let mut e = edit(&root, &service, &["email"]);
add(&mut e).unwrap();
e.commit().unwrap();
let registrations = fs::read_to_string(service.join("registrations.ts")).unwrap();
assert!(registrations.contains("EmailClient:"), "{registrations}");
assert!(registrations.contains("createEmailClient({"), "{registrations}");
assert!(registrations.contains("Lifetime.Scoped"));
assert!(registrations.contains("import { type EmailClient, OpenTelemetryCollector, createEmailClient } from \"@forklaunch/core/http\";"), "{registrations}");
assert!(registrations.contains("SchemaValidator, number, string, type } from \"@demo/core\""), "{registrations}");
assert!(!registrations.contains("SES"), "no vendor SDK or key");
assert!(!registrations.contains("API_KEY"));
let handler = fs::read_to_string(service.join("api/platformEvents/email.ts")).unwrap();
assert!(handler.contains("seen.has(event.id)"));
assert!(handler.contains("TODO: mark data.recipients undeliverable"));
let index = fs::read_to_string(service.join("api/platformEvents/index.ts")).unwrap();
assert!(index.contains("email: emailEvents"));
assert!(fs::read_to_string(service.join("server.ts")).unwrap().contains("platformEventsRouter"));
let _ = fs::remove_dir_all(&root);
}
#[test]
fn remove_takes_back_the_registration_and_its_imports() {
let (root, service) = service("remove");
let mut e = edit(&root, &service, &["email"]);
add(&mut e).unwrap();
e.commit().unwrap();
let mut e = edit(&root, &service, &[]);
remove(&mut e).unwrap();
e.remove_platform_events("email").unwrap();
e.commit().unwrap();
let registrations = fs::read_to_string(service.join("registrations.ts")).unwrap();
assert!(!registrations.contains("EmailClient"), "{registrations}");
assert_eq!(registrations, REGISTRATIONS);
assert!(!registrations.contains("createEmailClient"));
assert!(registrations.contains("import { OpenTelemetryCollector } from \"@forklaunch/core/http\";"));
assert!(registrations.contains("import { SchemaValidator, number, string } from \"@demo/core\";"), "{registrations}");
assert!(!service.join("api/platformEvents/email.ts").exists());
assert!(!service.join("api/routes/platformEvents.routes.ts").exists());
assert_eq!(fs::read_to_string(service.join("server.ts")).unwrap().trim_end(), SERVER);
let _ = fs::remove_dir_all(&root);
}
#[test]
fn named_imports_merge_and_unmerge_in_the_files_style() {
use super::super::remove_named_imports;
let text = "import { a } from '@x/y';\nimport {\n b,\n c\n} from \"@forklaunch/core/http\";\nconst z = 1;\n";
let added = add_named_imports(text, HTTP_SOURCE, HTTP_IMPORTS);
assert!(added.contains("import {\n type EmailClient,\n b,\n c,\n createEmailClient\n} from \"@forklaunch/core/http\";"), "{added}");
assert_eq!(add_named_imports(&added, HTTP_SOURCE, HTTP_IMPORTS), added, "idempotent");
assert_eq!(remove_named_imports(&added, HTTP_SOURCE, HTTP_IMPORTS), text);
let fresh = add_named_imports("import { a } from 'x';\nconst z = 1;\n", HTTP_SOURCE, &["createEmailClient"]);
assert_eq!(
fresh,
"import { createEmailClient } from '@forklaunch/core/http';\nimport { a } from 'x';\nconst z = 1;\n"
);
assert_eq!(
remove_named_imports(&fresh, HTTP_SOURCE, &["createEmailClient"]),
"import { a } from 'x';\nconst z = 1;\n"
);
}
#[test]
fn gateway_wiring_keeps_the_services_other_imports() {
let (root, service) = service("wiring");
let text = REGISTRATIONS.replace(
"import { Lifetime, createConfigInjector, getEnvVar } from \"@forklaunch/core/services\";",
"import {\n createConfigInjector,\n getEnvVar,\n Lifetime\n} from '@forklaunch/core/services';",
);
fs::write(service.join("registrations.ts"), text).unwrap();
let mut e = edit(&root, &service, &["email"]);
add(&mut e).unwrap();
e.ensure_gateway_wiring("email").unwrap();
e.commit().unwrap();
let registrations = fs::read_to_string(service.join("registrations.ts")).unwrap();
assert!(
registrations.contains("import {\n createConfigInjector,\n getEnvVar,\n Lifetime\n} from '@forklaunch/core/services';"),
"{registrations}"
);
assert!(registrations.contains("INSTANCE_HMAC_KEY:"));
let _ = fs::remove_dir_all(&root);
}
#[test]
fn email_is_a_registered_capability_that_receives_events() {
let c = super::super::find("email").expect("registered");
assert_eq!(c.resource_type, "email");
assert_eq!(c.registration_key, "EmailClient");
assert!(c.receives_events);
}
}