Merge pull request #174 from Schniz/schniz/support-email-triggering

feat: add support for email triggers

.changeset/thirty-ties-float.md

@@ -0,0 +1,15 @@
+---
+'@microlabs/otel-cf-workers': minor
+---
+
+add support for `email` handlers
+
+Example usage:
+
+```ts
+export default {
+  async email(message, env, ctx) {
+    // this is running in a trace!
+  },
+};
+```

README.md

@@ -266,7 +266,7 @@ One of the advantages of using Open Telemetry is that it makes it easier to do d
 
 Triggers:
 
-- [ ] Email (`handler.email`)
+- [x] Email (`handler.email`)
 - [x] HTTP (`handler.fetch`)
 - [x] Queue (`handler.queue`)
 - [x] Cron (`handler.scheduled`)

package.json

@@ -19,7 +19,7 @@
 	"scripts": {
 		"clean": "rimraf ./dist versions.json",
 		"format": "prettier --ignore-unknown --write .",
-		"build:src": "tsup src/index.ts --format cjs,esm --dts --clean --sourcemap",
+		"build:src": "tsup",
 		"build:versions": "pnpm version --json > versions.json",
 		"build": "run-s -l build:versions build:src",
 		"cs-version": "changeset version",

src/instrumentation/email.ts

@@ -0,0 +1,76 @@
+import { setConfig, type Initialiser } from '../config'
+import { wrap } from '../wrap'
+import { exportSpans, proxyExecutionContext } from './common'
+import { context as api_context, Exception, SpanKind, type SpanOptions, trace } from '@opentelemetry/api'
+import { instrumentEnv } from './env'
+import { versionAttributes } from './version'
+import {
+	ATTR_FAAS_TRIGGER,
+	ATTR_MESSAGING_DESTINATION_NAME,
+	ATTR_RPC_MESSAGE_ID,
+} from '@opentelemetry/semantic-conventions/incubating'
+
+type EmailHandler = EmailExportedHandler
+export type EmailHandlerArgs = Parameters<EmailHandler>
+
+export function createEmailHandler(emailFn: EmailHandler, initialiser: Initialiser): EmailHandler {
+	const emailHandler: ProxyHandler<EmailHandler> = {
+		async apply(target, _thisArg, argArray: Parameters<EmailHandler>): Promise<void> {
+			const [message, orig_env, orig_ctx] = argArray
+			const config = initialiser(orig_env as Record<string, unknown>, message)
+			const env = instrumentEnv(orig_env as Record<string, unknown>)
+			const { ctx, tracker } = proxyExecutionContext(orig_ctx)
+			const context = setConfig(config)
+
+			try {
+				const args: EmailHandlerArgs = [message, env, ctx]
+				return await api_context.with(context, executeEmailHandler, undefined, target, args)
+			} catch (error) {
+				throw error
+			} finally {
+				orig_ctx.waitUntil(exportSpans(tracker))
+			}
+		},
+	}
+	return wrap(emailFn, emailHandler)
+}
+
+/**
+ * Converts the message headers into a record ready to be injected
+ * as OpenTelemetry attributes
+ *
+ * @example
+ * ```ts
+ * const headers = new Headers({ "Subject": "Hello!", From: "[email protected]" })
+ * headerAttributes({ headers })
+ * // => {"email.header.Subject": "Hello!", "email.header.From": "[email protected]"}
+ * ```
+ */
+function headerAttributes(message: { headers: Headers }): Record<string, unknown> {
+	return Object.fromEntries([...message.headers].map(([key, value]) => [`email.header.${key}`, value] as const))
+}
+
+async function executeEmailHandler(emailFn: EmailHandler, [message, env, ctx]: EmailHandlerArgs): Promise<void> {
+	const tracer = trace.getTracer('emailHandler')
+	const options = {
+		attributes: {
+			[ATTR_FAAS_TRIGGER]: 'other',
+			[ATTR_RPC_MESSAGE_ID]: message.headers.get('Message-Id') ?? undefined,
+			[ATTR_MESSAGING_DESTINATION_NAME]: message.to,
+		},
+		kind: SpanKind.CONSUMER,
+	} satisfies SpanOptions
+	Object.assign(options.attributes!, headerAttributes(message), versionAttributes(env))
+	const promise = tracer.startActiveSpan(`emailHandler ${message.to}`, options, async (span) => {
+		try {
+			const result = await emailFn(message, env, ctx)
+			span.end()
+			return result
+		} catch (error) {
+			span.recordException(error as Exception)
+			span.end()
+			throw error
+		}
+	})
+	return promise
+}

src/sdk.ts

@@ -12,10 +12,12 @@ import { DOClass, instrumentDOClass } from './instrumentation/do.js'
 import { createScheduledHandler } from './instrumentation/scheduled.js'
 //@ts-ignore
 import * as versions from '../versions.json'
+import { createEmailHandler } from './instrumentation/email.js'
 
 type FetchHandler = ExportedHandlerFetchHandler<unknown, unknown>
 type ScheduledHandler = ExportedHandlerScheduledHandler<unknown>
 type QueueHandler = ExportedHandlerQueueHandler
+type EmailHandler = EmailExportedHandler
 
 export type ResolveConfigFn<Env = any> = (env: Env, trigger: Trigger) => TraceConfig
 export type ConfigurationOption = TraceConfig | ResolveConfigFn
@@ -106,6 +108,12 @@ export function instrument<E, Q, C>(
 		const queuer = unwrap(handler.queue) as QueueHandler
 		handler.queue = createQueueHandler(queuer, initialiser)
 	}
+
+	if (handler.email) {
+		const emailer = unwrap(handler.email) as EmailHandler
+		handler.email = createEmailHandler(emailer, initialiser)
+	}
+
 	return handler
 }
 

src/types.ts

@@ -73,4 +73,10 @@ export interface DOConstructorTrigger {
 	name?: string
 }
 
-export type Trigger = Request | MessageBatch | ScheduledController | DOConstructorTrigger | 'do-alarm'
+export type Trigger =
+	| Request
+	| MessageBatch
+	| ScheduledController
+	| DOConstructorTrigger
+	| 'do-alarm'
+	| ForwardableEmailMessage

tsup.config.ts

@@ -0,0 +1,9 @@
+import { defineConfig } from 'tsup'
+
+export default defineConfig({
+	entry: ['src/index.ts'],
+	format: ['cjs', 'esm'],
+	dts: true,
+	clean: true,
+	sourcemap: true,
+})