diff --git a/CHANGELOG.md b/CHANGELOG.md index 945e7d6..fb25f1c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added +- **Delivery callback diagnostics** on Admin → SES regions: per region, the SNS subscription state and endpoint, each configuration set's event destination, when the last callback arrived, and any callbacks rejected for an unknown topic, with one-click resubscribe to this site's callback URL + +### Fixed +- Sent counts in analytics no longer depend on SES callbacks: an email is counted when ByteSend hands it to SES, so totals stay accurate even if SNS notifications stop arriving + +### Changed +- **Node.js SDK 0.1.4**: package metadata points at the new `ByteSendApp/ByteSend` repository, and releases publish from it with provenance + --- ## [0.4.3] - 2026-10-19 diff --git a/apps/web/src/app/(dashboard)/admin/ses/callback-diagnostics.tsx b/apps/web/src/app/(dashboard)/admin/ses/callback-diagnostics.tsx new file mode 100644 index 0000000..3aed308 --- /dev/null +++ b/apps/web/src/app/(dashboard)/admin/ses/callback-diagnostics.tsx @@ -0,0 +1,145 @@ +"use client"; + +import { Button } from "@bytesend/ui/src/button"; +import Spinner from "@bytesend/ui/src/spinner"; +import { toast } from "@bytesend/ui/src/toaster"; +import { formatDistanceToNow } from "date-fns"; +import { RefreshCw } from "lucide-react"; +import { api } from "~/trpc/react"; +import { StatusPill, type StatusTone } from "../_components/status-pill"; + +const SUBSCRIPTION_TONE: Record = { + confirmed: { tone: "success", label: "Subscribed" }, + pending: { tone: "warning", label: "Pending confirmation" }, + missing: { tone: "destructive", label: "Not subscribed" }, + error: { tone: "destructive", label: "Lookup failed" }, +}; + +export default function CallbackDiagnostics() { + const utils = api.useUtils(); + const diagnosticsQuery = api.admin.getSesCallbackDiagnostics.useQuery(undefined, { + refetchOnWindowFocus: false, + }); + const resubscribe = api.admin.resubscribeSesCallback.useMutation({ + onSuccess: async (result) => { + toast.success("Resubscribed", { + description: `AWS will send a confirmation to ${result.callbackUrl}. Re-check in a few seconds.`, + }); + await Promise.all([utils.admin.getSesCallbackDiagnostics.invalidate(), utils.admin.getSesSettings.invalidate()]); + }, + onError: (error) => toast.error("Resubscribe failed", { description: error.message }), + }); + + const data = diagnosticsQuery.data; + + return ( +
+
+
+

Delivery callbacks

+

+ Delivered, bounced, opened and clicked statuses only arrive when SES can reach this site through SNS. +

+
+ +
+ + {data?.lastRejected && ( +
+

Callbacks are being rejected

+

+ SNS posted from {data.lastRejected.topicArn}{" "} + {formatDistanceToNow(new Date(data.lastRejected.at), { addSuffix: true })}, which doesn't match any + region below. Those events were dropped. +

+
+ )} + + {diagnosticsQuery.isLoading ? ( +
+ +
+ ) : diagnosticsQuery.error ? ( +

{diagnosticsQuery.error.message}

+ ) : ( +
+ {data?.regions.map((region) => { + const subscription = SUBSCRIPTION_TONE[region.subscription.state]!; + const healthy = region.problems.length === 0; + return ( +
+
+ {region.region} + + {healthy ? "Healthy" : `${region.problems.length} problem${region.problems.length === 1 ? "" : "s"}`} + +
+ +
+
+
SNS subscription
+
+ {subscription.label} +
+
+
+
Event destinations
+
+ {region.configSets.filter((configSet) => configSet.ok).length} / {region.configSets.length} OK +
+
+
+
Last callback
+
+ {region.lastReceivedAt + ? formatDistanceToNow(new Date(region.lastReceivedAt), { addSuffix: true }) + : "None recorded"} +
+
+
+
Callback URL
+
+ {region.callbackUrl} +
+
+
+ + {!healthy && ( +
    + {region.problems.map((problem) => ( +
  • + + {problem} +
    • + ))} +
+ )} + + {region.subscription.state !== "confirmed" || region.callbackUrl !== region.expectedCallbackUrl ? ( + + ) : null} +
+ ); + })} +
+ )} +
+ ); +} diff --git a/apps/web/src/app/(dashboard)/admin/ses/page.tsx b/apps/web/src/app/(dashboard)/admin/ses/page.tsx index 57f7f71..4fe080a 100644 --- a/apps/web/src/app/(dashboard)/admin/ses/page.tsx +++ b/apps/web/src/app/(dashboard)/admin/ses/page.tsx @@ -2,6 +2,7 @@ import { AdminPageHeader } from "../_components/admin-page-header"; import AddSesConfiguration from "./add-ses-configuration"; +import CallbackDiagnostics from "./callback-diagnostics"; import SesConfigurations from "./ses-configurations"; export default function AdminSesPage() { @@ -13,6 +14,7 @@ export default function AdminSesPage() { actions={} /> + ); } diff --git a/apps/web/src/app/api/ses_callback/route.ts b/apps/web/src/app/api/ses_callback/route.ts index 862a5be..60f19eb 100644 --- a/apps/web/src/app/api/ses_callback/route.ts +++ b/apps/web/src/app/api/ses_callback/route.ts @@ -1,6 +1,7 @@ import { db } from "~/server/db"; import { logger } from "~/server/logger/log"; import { parseSesHook, SesHookParser } from "~/server/service/ses-hook-parser"; +import { SesCallbackTracker } from "~/server/service/ses-callback-tracker"; import { SesSettingsService } from "~/server/service/ses-settings-service"; import { SnsNotificationMessage } from "~/types/aws-types"; @@ -29,9 +30,14 @@ export async function POST(req: Request) { logger.debug({ isEventValid }, "SES callback topic validation result"); if (!isEventValid) { + if (typeof data?.TopicArn === "string") { + await SesCallbackTracker.recordRejected(data.TopicArn); + } return Response.json({ data: "Event is not valid" }); } + await SesCallbackTracker.recordReceived(trustedTopicArn); + if (data.Type === "SubscriptionConfirmation") { return handleSubscription(data, trustedTopicArn); } diff --git a/apps/web/src/lib/constants/admin-audit.ts b/apps/web/src/lib/constants/admin-audit.ts index 1eca290..3d522cd 100644 --- a/apps/web/src/lib/constants/admin-audit.ts +++ b/apps/web/src/lib/constants/admin-audit.ts @@ -3,6 +3,7 @@ export const ADMIN_AUDIT_ACTIONS = { "ses.update": "Updated SES region", "ses.delete": "Deleted SES region", "ses.toggle": "Toggled SES region", + "ses.resubscribe": "Resubscribed SES callback", "user.beta.update": "Changed beta access", "user.admin.update": "Changed admin role", "user.ban": "Banned user", diff --git a/apps/web/src/server/api/routers/admin.ts b/apps/web/src/server/api/routers/admin.ts index eb2822d..b3800d9 100644 --- a/apps/web/src/server/api/routers/admin.ts +++ b/apps/web/src/server/api/routers/admin.ts @@ -17,6 +17,7 @@ import { createCheckoutSessionForTeam, type CheckoutPlan } from "~/server/billin import { TeamService } from "~/server/service/team-service"; import { AdminOverviewService } from "~/server/service/admin-overview-service"; import { AdminAuditService, diffFields } from "~/server/service/admin-audit-service"; +import { SesCallbackDiagnosticsService } from "~/server/service/ses-callback-diagnostics-service"; import { AdminDeliverabilityService } from "~/server/service/admin-deliverability-service"; import { AdminDetailService } from "~/server/service/admin-detail-service"; import { BlacklistService } from "~/server/service/blacklist-service"; @@ -458,6 +459,22 @@ export const adminRouter = createTRPCRouter({ return result; }), + getSesCallbackDiagnostics: adminProcedure.query(async () => SesCallbackDiagnosticsService.diagnose()), + + resubscribeSesCallback: adminProcedure + .input(z.object({ settingsId: z.string() })) + .mutation(async ({ ctx, input }) => { + const result = await SesCallbackDiagnosticsService.resubscribe(input.settingsId); + await AdminAuditService.record({ + actor: actorFrom(ctx), + action: "ses.resubscribe", + targetType: "ses", + targetId: input.settingsId, + metadata: { from: result.previousCallbackUrl, to: result.callbackUrl }, + }); + return result; + }), + getSetting: adminProcedure .input( z.object({ diff --git a/apps/web/src/server/aws/ses.ts b/apps/web/src/server/aws/ses.ts index bd3c613..51f7148 100644 --- a/apps/web/src/server/aws/ses.ts +++ b/apps/web/src/server/aws/ses.ts @@ -8,6 +8,7 @@ import { SendEmailCommand, CreateConfigurationSetEventDestinationCommand, CreateConfigurationSetCommand, + GetConfigurationSetEventDestinationsCommand, EventType, GetAccountCommand, CreateTenantResourceAssociationCommand, @@ -352,6 +353,19 @@ export async function addWebhookConfiguration( return response.$metadata.httpStatusCode === 200; } +export async function getConfigurationSetEventDestinations(configName: string, region: string) { + const sesClient = getSesClient(region); + const response = await sesClient.send( + new GetConfigurationSetEventDestinationsCommand({ ConfigurationSetName: configName }), + ); + return (response.EventDestinations ?? []).map((destination) => ({ + name: destination.Name ?? null, + enabled: destination.Enabled ?? false, + topicArn: destination.SnsDestination?.TopicArn ?? null, + eventTypes: destination.MatchingEventTypes ?? [], + })); +} + export async function deleteFromSesSuppressionList( email: string, region: string diff --git a/apps/web/src/server/aws/sns.ts b/apps/web/src/server/aws/sns.ts index 88487e7..d798593 100644 --- a/apps/web/src/server/aws/sns.ts +++ b/apps/web/src/server/aws/sns.ts @@ -3,6 +3,7 @@ import { CreateTopicCommand, SubscribeCommand, DeleteTopicCommand, + ListSubscriptionsByTopicCommand, } from "@aws-sdk/client-sns"; import { env } from "~/env"; @@ -48,3 +49,14 @@ export async function subscribeEndpoint( const data = await client.send(subscribeCommand); return data.SubscriptionArn; } + +export async function listTopicSubscriptions(topicArn: string, region: string) { + const client = getSnsClient(region); + const data = await client.send(new ListSubscriptionsByTopicCommand({ TopicArn: topicArn })); + return (data.Subscriptions ?? []).map((subscription) => ({ + endpoint: subscription.Endpoint ?? null, + protocol: subscription.Protocol ?? null, + // SNS reports "PendingConfirmation" here until the endpoint confirms the subscription. + subscriptionArn: subscription.SubscriptionArn ?? null, + })); +} diff --git a/apps/web/src/server/service/email-queue-service.ts b/apps/web/src/server/service/email-queue-service.ts index 7deb4dd..385cc59 100644 --- a/apps/web/src/server/service/email-queue-service.ts +++ b/apps/web/src/server/service/email-queue-service.ts @@ -476,11 +476,27 @@ async function executeEmail(job: QueueEmailJob) { data: { sesEmailId: messageId, text, attachments: null, headers: null }, }); - await db.email.updateMany({ + const { count: markedSent } = await db.email.updateMany({ where: { id: email.id, latestStatus: { in: ["QUEUED", "SCHEDULED"] } }, data: { latestStatus: "SENT" }, }); + // Counted here rather than from SES's Send event so sending totals don't depend on SNS + // callbacks arriving; markedSent guards against counting a retried job twice. + if (markedSent > 0) { + const usageKey = { + teamId: email.teamId, + domainId: email.domainId ?? 0, + date: new Date().toISOString().slice(0, 10), + type: email.campaignId ? ("MARKETING" as const) : ("TRANSACTIONAL" as const), + }; + await db.dailyEmailUsage.upsert({ + where: { teamId_domainId_date_type: usageKey }, + create: { ...usageKey, sent: 1 }, + update: { sent: { increment: 1 } }, + }); + } + } catch (error: any) { if (isSesTransientError(error)) { logger.warn( diff --git a/apps/web/src/server/service/ses-callback-diagnostics-service.ts b/apps/web/src/server/service/ses-callback-diagnostics-service.ts new file mode 100644 index 0000000..5ba9bec --- /dev/null +++ b/apps/web/src/server/service/ses-callback-diagnostics-service.ts @@ -0,0 +1,160 @@ +import type { SesSetting } from "@prisma/client"; +import { env } from "~/env"; +import { db } from "../db"; +import { getConfigurationSetEventDestinations } from "../aws/ses"; +import { listTopicSubscriptions, subscribeEndpoint } from "../aws/sns"; +import { logger } from "../logger/log"; +import { SesCallbackTracker, type RejectedCallback } from "./ses-callback-tracker"; +import { SesSettingsService } from "./ses-settings-service"; + +export type SubscriptionState = "confirmed" | "pending" | "missing" | "error"; + +export type ConfigSetDiagnosis = { + name: string; + ok: boolean; + detail: string; +}; + +export type RegionCallbackDiagnosis = { + settingId: string; + region: string; + topicArn: string | null; + callbackUrl: string; + expectedCallbackUrl: string; + subscription: { state: SubscriptionState; endpoint: string | null; detail: string }; + configSets: ConfigSetDiagnosis[]; + lastReceivedAt: string | null; + problems: string[]; +}; + +export type CallbackDiagnostics = { + regions: RegionCallbackDiagnosis[]; + lastRejected: RejectedCallback | null; +}; + +function expectedCallbackUrl() { + return `${env.NEXTAUTH_URL.replace(/\/$/, "")}/api/ses_callback`; +} + +function configSetNames(setting: SesSetting) { + return [setting.configGeneral, setting.configClick, setting.configOpen, setting.configFull].filter( + (name): name is string => Boolean(name), + ); +} + +async function diagnoseSubscription(setting: SesSetting): Promise { + if (!setting.topicArn) { + return { state: "missing", endpoint: null, detail: "No SNS topic is stored for this region" }; + } + try { + const subscriptions = await listTopicSubscriptions(setting.topicArn, setting.region); + const ours = subscriptions.find((subscription) => subscription.endpoint === setting.callbackUrl); + if (!ours) { + const others = subscriptions.map((subscription) => subscription.endpoint).filter(Boolean).join(", "); + return { + state: "missing", + endpoint: null, + detail: others + ? `The topic is subscribed to ${others}, not ${setting.callbackUrl}` + : "The topic has no subscriptions, so SES events go nowhere", + }; + } + if (ours.subscriptionArn === "PendingConfirmation") { + return { + state: "pending", + endpoint: ours.endpoint, + detail: "AWS is still waiting for this endpoint to confirm the subscription", + }; + } + return { state: "confirmed", endpoint: ours.endpoint, detail: "Confirmed" }; + } catch (error) { + logger.warn({ err: error, region: setting.region }, "[SesCallbackDiagnostics]: SNS lookup failed"); + return { + state: "error", + endpoint: null, + detail: error instanceof Error ? error.message : "Could not read the SNS topic", + }; + } +} + +async function diagnoseConfigSet(setting: SesSetting, name: string): Promise { + try { + const destinations = await getConfigurationSetEventDestinations(name, setting.region); + const destination = destinations.find((candidate) => candidate.topicArn === setting.topicArn); + if (!destination) { + return { name, ok: false, detail: "No event destination points at this region's SNS topic" }; + } + if (!destination.enabled) { + return { name, ok: false, detail: "The event destination is disabled" }; + } + if (!destination.eventTypes.includes("DELIVERY")) { + return { name, ok: false, detail: "The event destination doesn't publish DELIVERY events" }; + } + return { name, ok: true, detail: destination.eventTypes.join(", ") }; + } catch (error) { + return { name, ok: false, detail: error instanceof Error ? error.message : "Could not read the configuration set" }; + } +} + +async function diagnoseRegion(setting: SesSetting): Promise { + const expected = expectedCallbackUrl(); + const [subscription, configSets, lastReceivedAt] = await Promise.all([ + diagnoseSubscription(setting), + Promise.all(configSetNames(setting).map((name) => diagnoseConfigSet(setting, name))), + setting.topicArn ? SesCallbackTracker.lastReceived(setting.topicArn) : Promise.resolve(null), + ]); + + const problems: string[] = []; + if (setting.callbackUrl !== expected) { + problems.push(`The callback URL is ${setting.callbackUrl}, but this site is served from ${expected}`); + } + if (subscription.state !== "confirmed") problems.push(subscription.detail); + if (configSets.length === 0) problems.push("No configuration sets are stored for this region"); + for (const configSet of configSets.filter((candidate) => !candidate.ok)) { + problems.push(`${configSet.name}: ${configSet.detail}`); + } + + return { + settingId: setting.id, + region: setting.region, + topicArn: setting.topicArn, + callbackUrl: setting.callbackUrl, + expectedCallbackUrl: expected, + subscription, + configSets, + lastReceivedAt, + problems, + }; +} + +export const SesCallbackDiagnosticsService = { + async diagnose(): Promise { + const settings = await SesSettingsService.getAllSettings(); + const [regions, lastRejected] = await Promise.all([ + Promise.all(settings.map(diagnoseRegion)), + SesCallbackTracker.lastRejected(), + ]); + return { regions, lastRejected }; + }, + + /** + * Points the region's callback at this site's URL and (re)subscribes it to the SNS topic. + * SNS then sends a SubscriptionConfirmation, which /api/ses_callback confirms automatically. + */ + async resubscribe(settingId: string) { + const setting = await db.sesSetting.findUnique({ where: { id: settingId } }); + if (!setting?.topicArn) throw new Error("SES setting or SNS topic not found"); + + const callbackUrl = expectedCallbackUrl(); + if (setting.callbackUrl !== callbackUrl) { + await db.sesSetting.update({ + where: { id: setting.id }, + data: { callbackUrl, callbackSuccess: false }, + }); + await SesSettingsService.invalidateCache(); + } + + const subscriptionArn = await subscribeEndpoint(setting.topicArn, callbackUrl, setting.region); + return { previousCallbackUrl: setting.callbackUrl, callbackUrl, subscriptionArn: subscriptionArn ?? null }; + }, +}; diff --git a/apps/web/src/server/service/ses-callback-tracker.ts b/apps/web/src/server/service/ses-callback-tracker.ts new file mode 100644 index 0000000..bad6736 --- /dev/null +++ b/apps/web/src/server/service/ses-callback-tracker.ts @@ -0,0 +1,46 @@ +import { getRedis, redisKey } from "../redis"; +import { logger } from "../logger/log"; + +const RETENTION_SECONDS = 30 * 24 * 60 * 60; + +const receivedKey = (topicArn: string) => redisKey(`ses-callback:received:${topicArn}`); +const REJECTED_KEY = redisKey("ses-callback:rejected"); + +export type RejectedCallback = { topicArn: string; at: string }; + +// Tracking is best-effort: a Redis hiccup must never fail the SNS callback itself. +export const SesCallbackTracker = { + async recordReceived(topicArn: string) { + try { + await getRedis().set(receivedKey(topicArn), new Date().toISOString(), "EX", RETENTION_SECONDS); + } catch (error) { + logger.warn({ err: error }, "[SesCallbackTracker]: Could not record callback"); + } + }, + + async recordRejected(topicArn: string) { + try { + const value: RejectedCallback = { topicArn, at: new Date().toISOString() }; + await getRedis().set(REJECTED_KEY, JSON.stringify(value), "EX", RETENTION_SECONDS); + } catch (error) { + logger.warn({ err: error }, "[SesCallbackTracker]: Could not record rejected callback"); + } + }, + + async lastReceived(topicArn: string): Promise { + try { + return await getRedis().get(receivedKey(topicArn)); + } catch { + return null; + } + }, + + async lastRejected(): Promise { + try { + const raw = await getRedis().get(REJECTED_KEY); + return raw ? (JSON.parse(raw) as RejectedCallback) : null; + } catch { + return null; + } + }, +}; diff --git a/apps/web/src/server/service/ses-hook-parser.ts b/apps/web/src/server/service/ses-hook-parser.ts index aec7d95..ab3332e 100644 --- a/apps/web/src/server/service/ses-hook-parser.ts +++ b/apps/web/src/server/service/ses-hook-parser.ts @@ -209,7 +209,6 @@ export async function parseSesHook(data: SesEvent) { "CLICKED", "BOUNCED", "COMPLAINED", - "SENT", ].includes(mailStatus) ) { logger.info("Updating daily email usage"); @@ -234,7 +233,6 @@ export async function parseSesHook(data: SesEvent) { clicked: updateField === "clicked" ? 1 : 0, bounced: updateField === "bounced" ? 1 : 0, complained: updateField === "complained" ? 1 : 0, - sent: updateField === "sent" ? 1 : 0, hardBounced: isHardBounced ? 1 : 0, }, update: { diff --git a/packages/sdk/package.json b/packages/sdk/package.json index 61eceaa..7268771 100644 --- a/packages/sdk/package.json +++ b/packages/sdk/package.json @@ -1,6 +1,6 @@ { "name": "bytesend-js", - "version": "0.1.3", + "version": "0.1.4", "description": "Official JavaScript SDK for ByteSend, the email sending service built for developers.", "license": "AGPL-3.0-only", "repository": {