diff --git a/.env.example b/.env.example index 9b8af4b0f..cf9b311db 100644 --- a/.env.example +++ b/.env.example @@ -5,11 +5,22 @@ APP_VERSION=dev APP_URL=http://localhost:3000 AUTH_SECRET=replace-with-at-least-32-random-characters SIGNUPS_ENABLED=true -# HTTPS webhook used for verification-email delivery in staging/production. -# Receives JSON: { type, to, from, subject, text, verificationUrl }. -AUTH_EMAIL_WEBHOOK_URL= -AUTH_EMAIL_WEBHOOK_BEARER_TOKEN= -AUTH_EMAIL_FROM=Aiwa Creators +# Transactional email. Production uses implicit TLS SMTP (port 465). +# Web and worker services must load the same SMTP and sender settings. +# Never expose SMTP credentials through NEXT_PUBLIC_* variables. +MAIL_ENABLED=true +SMTP_HOST= +SMTP_PORT=465 +SMTP_USER= +SMTP_PASSWORD= +SMTP_EHLO_NAME=creator.aiwamediagroup.com +SMTP_POOL_MAX_CONNECTIONS=3 +SMTP_POOL_MAX_MESSAGES=100 +SMTP_CONNECTION_TIMEOUT_MS=10000 +SMTP_GREETING_TIMEOUT_MS=10000 +SMTP_SOCKET_TIMEOUT_MS=30000 +MAIL_SECURITY_FROM_ADDRESS=security@aiwamediagroup.com +MAIL_ROUTINE_FROM_ADDRESS=creator-tool@aiwamediagroup.com # MariaDB. Credentials below are for local Compose only. MARIADB_DATABASE=creator_platform diff --git a/README.md b/README.md index fb498b9a9..bce15e37b 100644 --- a/README.md +++ b/README.md @@ -436,3 +436,11 @@ Studio supports queued BytePlus image generation with wallet reservation, persistent PNG assets and authenticated downloads. See [the image-generation runbook](docs/image-generation.md) for deployment, recovery behavior and the live acceptance check. + + +## Transactional email + +Authentication, billing, membership, generation and report notifications use a +durable MariaDB outbox with BullMQ delivery and pooled TLS SMTP. See +[the transactional-email operations runbook](docs/operations/transactional-email.md) +for production configuration, sender identities, retry behavior and diagnostics. diff --git a/apps/web/package.json b/apps/web/package.json index 1318efab0..ef6461464 100644 --- a/apps/web/package.json +++ b/apps/web/package.json @@ -18,6 +18,7 @@ "@aiwa/credits": "workspace:*", "@aiwa/db": "workspace:*", "@aiwa/generation": "workspace:*", + "@aiwa/mail": "workspace:*", "@aiwa/organizations": "workspace:*", "@aiwa/payments": "workspace:*", "@aiwa/providers": "workspace:*", diff --git a/apps/web/src/app/admin/email/page.tsx b/apps/web/src/app/admin/email/page.tsx new file mode 100644 index 000000000..6749d7fa9 --- /dev/null +++ b/apps/web/src/app/admin/email/page.tsx @@ -0,0 +1,195 @@ +import { db } from "@aiwa/db"; +import { MailRetryButton } from "@/components/admin/mail-retry-button"; +import { MailVerifyButton } from "@/components/admin/mail-verify-button"; +import { StatusBadge } from "@/components/admin/primitives"; +import { Eyebrow } from "@/components/ui/creative"; +import { requirePlatformPermission } from "@/lib/request-auth"; + +function maskEmail(email: string): string { + const [local = "", domain] = email.split("@"); + if (!domain) return "••••"; + const visible = local.length <= 2 ? local.slice(0, 1) : local.slice(0, 2); + return `${visible}••••@${domain}`; +} + +function statusTone( + status: string, +): "success" | "warning" | "danger" | "neutral" { + if (status === "SENT") return "success"; + if (status === "FAILED") return "danger"; + if (status === "RETRY" || status === "SENDING") return "warning"; + return "neutral"; +} + +export default async function AdminEmailPage() { + await requirePlatformPermission("audit:read"); + + const [pending, retrying, failed, sent, recent] = await Promise.all([ + db.mailMessage.count({ + where: { status: { in: ["PENDING", "QUEUED", "SENDING"] } }, + }), + db.mailMessage.count({ where: { status: "RETRY" } }), + db.mailMessage.count({ where: { status: "FAILED" } }), + db.mailMessage.count({ where: { status: "SENT" } }), + db.mailMessage.findMany({ + orderBy: { createdAt: "desc" }, + take: 50, + select: { + id: true, + kind: true, + priority: true, + status: true, + template: true, + recipient: true, + attemptCount: true, + lastErrorCode: true, + lastErrorMessage: true, + createdAt: true, + sentAt: true, + nextAttemptAt: true, + }, + }), + ]); + + const cards = [ + ["Queued", pending], + ["Retrying", retrying], + ["Failed", failed], + ["Sent", sent], + ] as const; + + return ( +
+ Operations +
+

+ Email delivery +

+

+ Durable transactional-mail outbox and delivery diagnostics. Security + content is never displayed here. +

+
+ +
+
+ +
+ {cards.map(([label, value]) => ( +
+

+ {label} +

+

+ {value.toLocaleString("en-US")} +

+
+ ))} +
+ +
+
+

+ Recent deliveries +

+

+ Latest 50 messages. Recipients are masked and mail bodies are + excluded. +

+
+
+ + + + + + + + + + + + + + + + {recent.map((mail) => ( + + + + + + + + + + + + ))} + {recent.length === 0 ? ( + + + + ) : null} + +
CreatedRecipientChannelTemplateStatusAttemptsNext / sentLast errorAction
+ {mail.createdAt.toLocaleString("en-OM")} + + {maskEmail(mail.recipient)} + + + {mail.kind} + + + {mail.priority} + + + {mail.template} + + + {mail.status} + + + {mail.attemptCount} + + {(mail.sentAt ?? mail.nextAttemptAt)?.toLocaleString( + "en-OM", + ) ?? "—"} + + {mail.lastErrorCode ? ( + <> + + {mail.lastErrorCode} + + {mail.lastErrorMessage ? ( + + {mail.lastErrorMessage} + + ) : null} + + ) : ( + "—" + )} + + {mail.kind === "ROUTINE" && + ["FAILED", "RETRY"].includes(mail.status) ? ( + + ) : ( + + — + + )} +
+ No transactional mail has been queued yet. +
+
+
+
+ ); +} diff --git a/apps/web/src/app/admin/layout.tsx b/apps/web/src/app/admin/layout.tsx index 73faaa078..39b18b158 100644 --- a/apps/web/src/app/admin/layout.tsx +++ b/apps/web/src/app/admin/layout.tsx @@ -55,6 +55,12 @@ const navigation: readonly (AdminNavigationItem & { icon: "assets", permission: "audit:read", }, + { + href: "/admin/email", + label: "Email delivery", + icon: "activity", + permission: "audit:read", + }, ]; export default async function AdminLayout({ diff --git a/apps/web/src/app/api/account/notification-preferences/route.ts b/apps/web/src/app/api/account/notification-preferences/route.ts new file mode 100644 index 000000000..b80f7fd61 --- /dev/null +++ b/apps/web/src/app/api/account/notification-preferences/route.ts @@ -0,0 +1,83 @@ +import { db } from "@aiwa/db"; +import { NextResponse } from "next/server"; +import { z } from "zod"; +import { getRequestSession } from "@/lib/request-auth"; +import { hasTrustedMutationOrigin } from "@/lib/request-security"; + +const schema = z.object({ + generationCompleted: z.boolean(), + generationFailed: z.boolean(), + reports: z.boolean(), +}); + +export async function GET(request: Request) { + const session = await getRequestSession(request.headers); + if (!session) { + return NextResponse.json( + { error: "Authentication required." }, + { status: 401 }, + ); + } + + const preference = await db.notificationPreference.findUnique({ + where: { userId: session.user.id }, + select: { + generationCompleted: true, + generationFailed: true, + reports: true, + }, + }); + + return NextResponse.json( + preference ?? { + generationCompleted: true, + generationFailed: true, + reports: true, + }, + ); +} + +export async function PATCH(request: Request) { + if (!hasTrustedMutationOrigin(request)) { + return NextResponse.json({ error: "Origin not allowed." }, { status: 403 }); + } + + const session = await getRequestSession(request.headers); + if (!session) { + return NextResponse.json( + { error: "Authentication required." }, + { status: 401 }, + ); + } + + const parsed = schema.safeParse(await request.json().catch(() => null)); + if (!parsed.success) { + return NextResponse.json( + { error: "Invalid notification preferences." }, + { status: 400 }, + ); + } + + const preference = await db.notificationPreference.upsert({ + where: { userId: session.user.id }, + create: { userId: session.user.id, ...parsed.data }, + update: parsed.data, + select: { + generationCompleted: true, + generationFailed: true, + reports: true, + }, + }); + + await db.auditEvent.create({ + data: { + actorUserId: session.user.id, + action: "user.notification_preferences_updated", + targetType: "User", + targetId: session.user.id, + metadata: preference, + }, + }); + + return NextResponse.json(preference); +} diff --git a/apps/web/src/app/api/admin/email/[mailId]/retry/route.ts b/apps/web/src/app/api/admin/email/[mailId]/retry/route.ts new file mode 100644 index 000000000..dd6f0cd57 --- /dev/null +++ b/apps/web/src/app/api/admin/email/[mailId]/retry/route.ts @@ -0,0 +1,86 @@ +import { hasPlatformPermission } from "@aiwa/authz"; +import { db } from "@aiwa/db"; +import { cuidSchema } from "@aiwa/validation"; +import { revalidatePath } from "next/cache"; +import { NextResponse } from "next/server"; +import { getRequestSession } from "@/lib/request-auth"; +import { hasTrustedMutationOrigin } from "@/lib/request-security"; + +export async function POST( + request: Request, + { params }: { params: Promise<{ mailId: string }> }, +) { + if (!hasTrustedMutationOrigin(request)) { + return NextResponse.json({ error: "Origin not allowed." }, { status: 403 }); + } + + const session = await getRequestSession(request.headers); + if (!session) { + return NextResponse.json( + { error: "Authentication required." }, + { status: 401 }, + ); + } + if (!hasPlatformPermission(session.user.platformRole, "jobs:manage")) { + return NextResponse.json({ error: "Permission denied." }, { status: 403 }); + } + + const { mailId } = await params; + if (!cuidSchema.safeParse(mailId).success) { + return NextResponse.json( + { error: "Mail delivery not found." }, + { status: 404 }, + ); + } + + const current = await db.mailMessage.findUnique({ + where: { id: mailId }, + select: { id: true, kind: true, status: true, attemptCount: true }, + }); + if (!current) { + return NextResponse.json( + { error: "Mail delivery not found." }, + { status: 404 }, + ); + } + if (current.kind !== "ROUTINE") { + return NextResponse.json( + { error: "Security mail must be regenerated from its source workflow." }, + { status: 409 }, + ); + } + if (!["FAILED", "RETRY"].includes(current.status)) { + return NextResponse.json( + { error: "Only failed or retrying routine mail can be requeued." }, + { status: 409 }, + ); + } + + const updated = await db.mailMessage.update({ + where: { id: current.id }, + data: { + status: "PENDING", + nextAttemptAt: null, + failedAt: null, + lastErrorCode: null, + lastErrorMessage: null, + }, + select: { id: true, status: true }, + }); + + await db.auditEvent.create({ + data: { + actorUserId: session.user.id, + action: "mail.routine_requeued", + targetType: "MailMessage", + targetId: current.id, + metadata: { + previousStatus: current.status, + previousAttemptCount: current.attemptCount, + }, + }, + }); + + revalidatePath("/admin/email"); + return NextResponse.json(updated); +} diff --git a/apps/web/src/app/api/admin/email/verify/route.ts b/apps/web/src/app/api/admin/email/verify/route.ts new file mode 100644 index 000000000..827aaa5f5 --- /dev/null +++ b/apps/web/src/app/api/admin/email/verify/route.ts @@ -0,0 +1,66 @@ +import { hasPlatformPermission } from "@aiwa/authz"; +import { parseServerEnv } from "@aiwa/config"; +import { db } from "@aiwa/db"; +import { classifySmtpFailure, verifySmtpTransport } from "@aiwa/mail/transport"; +import { NextResponse } from "next/server"; +import { getRequestSession } from "@/lib/request-auth"; +import { hasTrustedMutationOrigin } from "@/lib/request-security"; + +export async function POST(request: Request) { + if (!hasTrustedMutationOrigin(request)) { + return NextResponse.json({ error: "Origin not allowed." }, { status: 403 }); + } + + const session = await getRequestSession(request.headers); + if (!session) { + return NextResponse.json( + { error: "Authentication required." }, + { status: 401 }, + ); + } + if (!hasPlatformPermission(session.user.platformRole, "jobs:manage")) { + return NextResponse.json({ error: "Permission denied." }, { status: 403 }); + } + + const env = parseServerEnv(); + if (!env.MAIL_ENABLED) { + return NextResponse.json( + { ok: false, code: "MAIL_DISABLED" }, + { status: 409 }, + ); + } + + try { + await verifySmtpTransport(env); + await db.auditEvent.create({ + data: { + actorUserId: session.user.id, + action: "mail.smtp_verified", + targetType: "MailTransport", + metadata: { + host: env.SMTP_HOST, + port: env.SMTP_PORT, + }, + }, + }); + return NextResponse.json({ ok: true }); + } catch (error) { + const failure = classifySmtpFailure(error); + await db.auditEvent.create({ + data: { + actorUserId: session.user.id, + action: "mail.smtp_verification_failed", + targetType: "MailTransport", + metadata: { + host: env.SMTP_HOST, + port: env.SMTP_PORT, + code: failure.code, + }, + }, + }); + return NextResponse.json( + { ok: false, code: failure.code }, + { status: 503 }, + ); + } +} diff --git a/apps/web/src/app/api/health/route.ts b/apps/web/src/app/api/health/route.ts index eeb2ef481..d9c47bdc1 100644 --- a/apps/web/src/app/api/health/route.ts +++ b/apps/web/src/app/api/health/route.ts @@ -14,6 +14,13 @@ export function GET(): Response { version: process.env.APP_VERSION ?? "dev", mediaConfigured: isBytePlusMediaConfigured(), voiceConfigured: isBytePlusVoiceConfigured(), + mailConfigured: Boolean( + process.env.SMTP_HOST && + process.env.SMTP_USER && + process.env.SMTP_PASSWORD && + process.env.MAIL_SECURITY_FROM_ADDRESS && + process.env.MAIL_ROUTINE_FROM_ADDRESS, + ), timestamp: new Date().toISOString(), }, { diff --git a/apps/web/src/app/settings/notifications/page.tsx b/apps/web/src/app/settings/notifications/page.tsx new file mode 100644 index 000000000..a6b028b5b --- /dev/null +++ b/apps/web/src/app/settings/notifications/page.tsx @@ -0,0 +1,62 @@ +import { db } from "@aiwa/db"; +import Link from "next/link"; +import { NotificationPreferencesForm } from "@/components/settings/notification-preferences-form"; +import { Button } from "@/components/ui/button"; +import { Eyebrow } from "@/components/ui/creative"; +import { requireRequestSession } from "@/lib/request-auth"; + +export default async function NotificationSettingsPage() { + const session = await requireRequestSession("/settings/notifications"); + const preference = await db.notificationPreference.findUnique({ + where: { userId: session.user.id }, + select: { + generationCompleted: true, + generationFailed: true, + reports: true, + }, + }); + + return ( +
+
+ Preferences +
+
+

+ Email notifications +

+

+ Choose which routine Creator Platform messages arrive by email. +

+
+ +
+ +
+

+ Security messages always stay on +

+

+ Email verification, password and MFA changes, billing events, and + team membership notices cannot be disabled because they protect the + account and workspace. +

+
+ +
+ +
+
+
+ ); +} diff --git a/apps/web/src/components/admin/mail-retry-button.tsx b/apps/web/src/components/admin/mail-retry-button.tsx new file mode 100644 index 000000000..137ae4bd9 --- /dev/null +++ b/apps/web/src/components/admin/mail-retry-button.tsx @@ -0,0 +1,37 @@ +"use client"; + +import { useRouter } from "next/navigation"; +import { useState } from "react"; +import { Button } from "@/components/ui/button"; + +export function MailRetryButton({ mailId }: { mailId: string }) { + const router = useRouter(); + const [busy, setBusy] = useState(false); + + async function retry() { + setBusy(true); + try { + const response = await fetch( + `/api/admin/email/${encodeURIComponent(mailId)}/retry`, + { + method: "POST", + }, + ); + if (response.ok) router.refresh(); + } finally { + setBusy(false); + } + } + + return ( + + ); +} diff --git a/apps/web/src/components/admin/mail-verify-button.tsx b/apps/web/src/components/admin/mail-verify-button.tsx new file mode 100644 index 000000000..a0774e81e --- /dev/null +++ b/apps/web/src/components/admin/mail-verify-button.tsx @@ -0,0 +1,60 @@ +"use client"; + +import { useState } from "react"; +import { Button } from "@/components/ui/button"; + +export function MailVerifyButton() { + const [state, setState] = useState< + | { status: "idle" } + | { status: "checking" } + | { status: "ok" } + | { status: "error"; code: string } + >({ status: "idle" }); + + async function verify() { + setState({ status: "checking" }); + try { + const response = await fetch("/api/admin/email/verify", { + method: "POST", + }); + const body = (await response.json().catch(() => null)) as { + ok?: boolean; + code?: string; + } | null; + + if (response.ok && body?.ok) { + setState({ status: "ok" }); + return; + } + setState({ + status: "error", + code: body?.code ?? "SMTP_VERIFY_FAILED", + }); + } catch { + setState({ status: "error", code: "NETWORK_ERROR" }); + } + } + + return ( +
+ + {state.status === "ok" ? ( + + SMTP TLS and authentication verified. + + ) : state.status === "error" ? ( + + {state.code} + + ) : null} +
+ ); +} diff --git a/apps/web/src/components/settings/notification-preferences-form.tsx b/apps/web/src/components/settings/notification-preferences-form.tsx new file mode 100644 index 000000000..9dce53aed --- /dev/null +++ b/apps/web/src/components/settings/notification-preferences-form.tsx @@ -0,0 +1,100 @@ +"use client"; + +import { useState } from "react"; +import { Button } from "@/components/ui/button"; + +type Preferences = { + generationCompleted: boolean; + generationFailed: boolean; + reports: boolean; +}; + +export function NotificationPreferencesForm({ + initial, +}: { + initial: Preferences; +}) { + const [value, setValue] = useState(initial); + const [state, setState] = useState<"idle" | "saving" | "saved" | "error">( + "idle", + ); + + async function save() { + setState("saving"); + const response = await fetch("/api/account/notification-preferences", { + method: "PATCH", + headers: { "content-type": "application/json" }, + body: JSON.stringify(value), + }); + setState(response.ok ? "saved" : "error"); + } + + const options = [ + { + key: "generationCompleted" as const, + title: "Generation completed", + description: + "Receive a link when an image, video, or voice generation is ready.", + }, + { + key: "generationFailed" as const, + title: "Generation failed", + description: + "Receive a message when a generation cannot be completed and needs attention.", + }, + { + key: "reports" as const, + title: "Reports and summaries", + description: + "Receive routine platform reports and summaries when those features are enabled.", + }, + ]; + + return ( +
+ {options.map((option) => ( + + ))} + +
+ + {state === "saved" ? ( + Preferences saved. + ) : null} + {state === "error" ? ( + + Could not save preferences. + + ) : null} +
+
+ ); +} diff --git a/apps/web/src/lib/auth.ts b/apps/web/src/lib/auth.ts index 3228ad9a1..08e9ae507 100644 --- a/apps/web/src/lib/auth.ts +++ b/apps/web/src/lib/auth.ts @@ -1,8 +1,16 @@ +import { createHash } from "node:crypto"; import { platformRoles } from "@aiwa/authz"; import { parseServerEnv } from "@aiwa/config"; import { db } from "@aiwa/db"; +import { + enqueueMail, + passwordResetEmail, + securityEventEmail, + verificationEmail, +} from "@aiwa/mail"; import { betterAuth } from "better-auth"; import { prismaAdapter } from "better-auth/adapters/prisma"; +import { createAuthMiddleware } from "better-auth/api"; import { twoFactor } from "better-auth/plugins"; const env = parseServerEnv(); @@ -10,43 +18,18 @@ const env = parseServerEnv(); async function deliverVerificationEmail(input: { email: string; verificationUrl: string; + userId?: string; }): Promise { - if (!env.AUTH_EMAIL_WEBHOOK_URL) { - if (env.NODE_ENV !== "production") { - console.info( - `[EmailVerification] Verification link for ${input.email}: ${input.verificationUrl}`, - ); - return; - } - throw new Error( - "Authentication email delivery is not configured. Set AUTH_EMAIL_WEBHOOK_URL.", - ); - } - - const headers: Record = { - "content-type": "application/json", - }; - if (env.AUTH_EMAIL_WEBHOOK_BEARER_TOKEN) { - headers.authorization = `Bearer ${env.AUTH_EMAIL_WEBHOOK_BEARER_TOKEN}`; - } - - const response = await fetch(env.AUTH_EMAIL_WEBHOOK_URL, { - method: "POST", - headers, - body: JSON.stringify({ - type: "email_verification", + await enqueueMail( + verificationEmail({ to: input.email, - from: env.AUTH_EMAIL_FROM ?? "Aiwa Creators", - subject: "Verify your Aiwa Creators email", - text: `Verify your email address to activate workspace access: ${input.verificationUrl}`, verificationUrl: input.verificationUrl, + idempotencyKey: `verify:${createHash("sha256") + .update(input.verificationUrl) + .digest("hex")}`, + userId: input.userId, }), - }); - if (!response.ok) { - throw new Error( - `Authentication email delivery failed with status ${response.status}.`, - ); - } + ); } export const auth = betterAuth({ @@ -70,6 +53,7 @@ export const auth = betterAuth({ await deliverVerificationEmail({ email: user.email, verificationUrl: url, + userId: user.id, }); }, }, @@ -81,6 +65,29 @@ export const auth = betterAuth({ maxPasswordLength: 128, autoSignIn: false, revokeSessionsOnPasswordReset: true, + resetPasswordTokenExpiresIn: 60 * 60, + sendResetPassword: async ({ user, url, token }) => { + await enqueueMail( + passwordResetEmail({ + to: user.email, + resetUrl: url, + userId: user.id, + idempotencyKey: `password-reset:${createHash("sha256") + .update(token) + .digest("hex")}`, + }), + ); + }, + onPasswordReset: async ({ user }) => { + await enqueueMail( + securityEventEmail({ + to: user.email, + userId: user.id, + event: "PASSWORD_RESET", + idempotencyKey: `security:password-reset:${user.id}:${Date.now()}`, + }), + ); + }, }, account: { accountLinking: { @@ -138,7 +145,90 @@ export const auth = betterAuth({ validateSchema: true, }, }, + hooks: { + after: createAuthMiddleware(async (ctx) => { + if (ctx.path !== "/two-factor/generate-backup-codes") return; + + const session = ctx.context.session; + const userId = session?.user?.id; + if (!userId) return; + + const [user, factor] = await Promise.all([ + db.user.findUnique({ + where: { id: userId }, + select: { email: true }, + }), + db.twoFactor.findFirst({ + where: { userId }, + orderBy: { updatedAt: "desc" }, + select: { updatedAt: true }, + }), + ]); + if (!user || !factor) return; + + await enqueueMail( + securityEventEmail({ + to: user.email, + userId, + event: "BACKUP_CODES_REGENERATED", + idempotencyKey: `security:backup-codes-regenerated:${userId}:${factor.updatedAt.getTime()}`, + }), + ); + }), + }, databaseHooks: { + user: { + update: { + after: async (updatedUser, ctx) => { + if ( + ctx?.path !== "/two-factor/verify-totp" && + ctx?.path !== "/two-factor/disable" + ) { + return; + } + + const user = await db.user.findUnique({ + where: { id: updatedUser.id }, + select: { + id: true, + email: true, + twoFactorEnabled: true, + updatedAt: true, + }, + }); + if (!user) return; + + if (ctx.path === "/two-factor/verify-totp" && user.twoFactorEnabled) { + const factor = await db.twoFactor.findFirst({ + where: { userId: user.id, verified: true }, + orderBy: { updatedAt: "desc" }, + select: { updatedAt: true }, + }); + if (!factor) return; + await enqueueMail( + securityEventEmail({ + to: user.email, + userId: user.id, + event: "MFA_ENABLED", + idempotencyKey: `security:mfa-enabled:${user.id}:${factor.updatedAt.getTime()}`, + }), + ); + return; + } + + if (ctx.path === "/two-factor/disable" && !user.twoFactorEnabled) { + await enqueueMail( + securityEventEmail({ + to: user.email, + userId: user.id, + event: "MFA_DISABLED", + idempotencyKey: `security:mfa-disabled:${user.id}:${user.updatedAt.getTime()}`, + }), + ); + } + }, + }, + }, session: { create: { before: async (session) => { diff --git a/apps/worker/package.json b/apps/worker/package.json index ecdefd318..c9bccd79a 100644 --- a/apps/worker/package.json +++ b/apps/worker/package.json @@ -21,6 +21,7 @@ "bullmq": "6.3.5", "ioredis": "^5.8.1", "@aiwa/generation": "workspace:*", + "@aiwa/mail": "workspace:*", "@aiwa/db": "workspace:*", "@prisma/client": "6.19.3" }, diff --git a/apps/worker/src/index.ts b/apps/worker/src/index.ts index 8fd3ce420..89ad3ceec 100644 --- a/apps/worker/src/index.ts +++ b/apps/worker/src/index.ts @@ -6,6 +6,11 @@ import { processVideoSubmitJob, processVoiceJob, } from "@aiwa/generation/process"; +import { + closeSmtpTransport, + processMailMessage, + recoverStaleMailDeliveries, +} from "@aiwa/mail/transport"; import { createBytePlusProvider } from "@aiwa/providers/byteplus"; import { createNvidiaProvider } from "@aiwa/providers/nvidia"; import { Queue, Worker } from "bullmq"; @@ -60,6 +65,36 @@ maintenanceWorker.on("failed", (job, error) => { }); }); +const mailQueue = new Queue("mail", { + connection: redis, + prefix: "aiwa", +}); + +const mailWorker = new Worker( + "mail", + async (job) => { + if (!env.MAIL_ENABLED) return; + if ( + typeof job.data.mailId !== "string" || + typeof job.data.attemptNumber !== "number" + ) { + throw new Error("Invalid mail queue payload"); + } + await processMailMessage(env, job.data.mailId, job.data.attemptNumber, 5); + }, + { connection: redis, prefix: "aiwa", concurrency: 3 }, +); + +mailWorker.on("error", () => log("error", "Mail queue connection failed")); +mailWorker.on("failed", (job, error) => + log("error", "Mail delivery attempt failed", { + mailId: job?.data?.mailId, + attemptsMade: job?.attemptsMade, + errorName: error.name, + errorMessage: error.message, + }), +); + const generationQueue = new Queue("generation", { connection: redis, prefix: "aiwa", @@ -158,6 +193,76 @@ reasoningWorker.on("failed", (job, error) => ); let isShuttingDown = false; +let mailDispatching = false; +async function dispatchMail() { + if (isShuttingDown || mailDispatching || !env.MAIL_ENABLED) return; + mailDispatching = true; + try { + const now = new Date(); + const recovered = await recoverStaleMailDeliveries( + new Date(now.getTime() - 10 * 60 * 1000), + ); + if (recovered > 0) { + log("error", "Recovered stale interrupted mail deliveries", { + recovered, + }); + } + + const rows = await db.mailMessage.findMany({ + where: { + OR: [ + { status: "PENDING" }, + { status: "QUEUED" }, + { status: "RETRY", nextAttemptAt: { lte: now } }, + ], + }, + orderBy: [{ priority: "asc" }, { createdAt: "asc" }], + take: 100, + }); + + for (const row of rows) { + if (isShuttingDown) break; + const jobId = `mail:${row.id}`; + const existing = await mailQueue.getJob(jobId); + if (existing) { + const state = await existing.getState(); + if (["failed", "completed"].includes(state)) { + await existing.remove(); + } else { + if (row.status !== "QUEUED") { + await db.mailMessage.update({ + where: { id: row.id }, + data: { status: "QUEUED", queuedAt: new Date() }, + }); + } + continue; + } + } + await mailQueue.add( + "deliver", + { mailId: row.id, attemptNumber: row.attemptCount + 1 }, + { + jobId, + attempts: 1, + removeOnComplete: true, + removeOnFail: 100, + }, + ); + await db.mailMessage.update({ + where: { id: row.id }, + data: { status: "QUEUED", queuedAt: new Date() }, + }); + } + } catch (error) { + log("error", "Mail dispatch unavailable; outbox rows retained", { + errorName: error instanceof Error ? error.name : "UnknownError", + errorMessage: error instanceof Error ? error.message : "Unknown", + }); + } finally { + mailDispatching = false; + } +} + let generationDispatching = false; async function dispatchGeneration() { if (isShuttingDown || generationDispatching || !bytePlusProvider) return; @@ -285,6 +390,7 @@ async function dispatchReasoning() { } } +const mailDispatchTimer = setInterval(() => void dispatchMail(), 5_000); const generationDispatchTimer = setInterval( () => void dispatchGeneration(), 10_000, @@ -293,17 +399,20 @@ const reasoningDispatchTimer = setInterval( () => void dispatchReasoning(), 2_000, ); +void dispatchMail(); void dispatchGeneration(); void dispatchReasoning(); async function shutdown(signal: NodeJS.Signals): Promise { log("info", "worker shutting down", { signal }); isShuttingDown = true; + clearInterval(mailDispatchTimer); clearInterval(generationDispatchTimer); clearInterval(reasoningDispatchTimer); // Stop accepting new jobs from Redis queues immediately await Promise.allSettled([ + mailWorker.pause(true), generationWorker.pause(true), reasoningWorker.pause(true), ]); @@ -318,11 +427,14 @@ async function shutdown(signal: NodeJS.Signals): Promise { ); const drainTimeoutMs = providerDeadlineMs + 120_000 + 30_000; const drainPromise = Promise.all([ + mailWorker.close(), generationWorker.close(), reasoningWorker.close(), + mailQueue.close(), generationQueue.close(), reasoningQueue.close(), maintenanceWorker.close(), + closeSmtpTransport(), ]); let timeoutHandle: NodeJS.Timeout | undefined; @@ -368,8 +480,9 @@ for (const signal of ["SIGINT", "SIGTERM"] as const) { } log("info", "worker started", { - queues: ["maintenance", "generation", "reasoning"], + queues: ["maintenance", "mail", "generation", "reasoning"], environment: env.APP_ENV, bytePlusConfigured: Boolean(bytePlusProvider), nvidiaConfigured: Boolean(nvidiaProvider), + mailEnabled: env.MAIL_ENABLED, }); diff --git a/docs/operations/transactional-email.md b/docs/operations/transactional-email.md new file mode 100644 index 000000000..cfcec67f4 --- /dev/null +++ b/docs/operations/transactional-email.md @@ -0,0 +1,76 @@ +# Transactional email operations + +Aiwa Creators uses a durable MariaDB outbox plus the BullMQ worker for all +outbound email. + +## Sender identities + +- `security@aiwamediagroup.com`: verification, password and MFA events, billing, + invitations, and team membership. +- `creator-tool@aiwamediagroup.com`: routine generation, report, and activity + notifications. + +Security mail is mandatory. Routine categories can be disabled by users at +`/settings/notifications`. + +## Production SMTP configuration + +Use implicit TLS SMTP on port 465. Certificate verification remains enabled. + +```dotenv +MAIL_ENABLED=true +SMTP_HOST=mail.example.com +SMTP_PORT=465 +SMTP_USER= +SMTP_PASSWORD= +SMTP_EHLO_NAME=creator.aiwamediagroup.com +SMTP_POOL_MAX_CONNECTIONS=3 +SMTP_POOL_MAX_MESSAGES=100 +SMTP_CONNECTION_TIMEOUT_MS=10000 +SMTP_GREETING_TIMEOUT_MS=10000 +SMTP_SOCKET_TIMEOUT_MS=30000 +MAIL_SECURITY_FROM_ADDRESS=security@aiwamediagroup.com +MAIL_ROUTINE_FROM_ADDRESS=creator-tool@aiwamediagroup.com +``` + +When `APP_ENV=production` and `MAIL_ENABLED=true`, startup fails closed unless +the SMTP host, username, password, both sender addresses, and port 465 are +explicitly configured. This prevents a deployment from silently accepting +security mail into an undeliverable queue. + +Apply database migrations before restarting services: + +```bash +pnpm --filter @aiwa/db migrate:deploy +sudo systemctl restart creator-web.service creator-worker.service +``` + +## Reliability model + +Business operations persist an outbox row before delivery. Redis is only the +dispatcher, so Redis or SMTP downtime does not lose queued mail. Temporary +delivery failures retry with bounded backoff. A worker interruption that leaves +a message in `SENDING` is recovered after ten minutes and retried using the same +RFC Message-ID, reducing duplicate-delivery risk. Permanent SMTP failures remain +visible in the admin Email delivery page. + +Security-message bodies are redacted after successful delivery. Failed security +messages are not generically replayable because reset or invitation tokens may +have expired; regenerate them from the original workflow instead. Failed routine +messages can be explicitly requeued by an authorized operator. + +## Operational checks + +- `/api/health` exposes whether SMTP credentials and both sender identities are + configured. +- `/admin/email` shows queue state, delivery counts, masked recipients, + attempts, and sanitized failure diagnostics. +- **Verify SMTP** on `/admin/email` performs a live TLS/authentication handshake + without sending mail. The result is audited and only a sanitized failure code + is returned to the browser. +- SMTP credentials, reset links, verification links, and MFA secrets must never + be logged. +- Use the admin SMTP verification control after every credential or mail-server + change. +- SPF, DKIM, and DMARC should be configured for the sending domain before + production launch. diff --git a/docs/security/authentication-and-rbac.md b/docs/security/authentication-and-rbac.md index 83a5afdfc..3b5467b19 100644 --- a/docs/security/authentication-and-rbac.md +++ b/docs/security/authentication-and-rbac.md @@ -7,8 +7,11 @@ login, phone login, passkeys, and magic links are intentionally not configured. Better Auth owns credentials and database sessions; Aiwa owns the organization and authorization model. -Email verification and password-reset email delivery remain disabled until a -transactional email provider and verified sender domain are configured. The +Email verification is delivered through the durable transactional-mail outbox. +Security, billing, and team-membership messages use +`security@aiwamediagroup.com`; routine user activity uses +`creator-tool@aiwamediagroup.com`. SMTP delivery is performed only by the worker +over implicit TLS, while MariaDB remains the delivery source of truth. The password policy requires 12 to 128 characters. ## Signup and organization onboarding @@ -88,3 +91,23 @@ Apply the database migration before starting the new web image: ```bash pnpm --filter @aiwa/db migrate:deploy ``` + +## Transactional email operations + +Production mail requires `MAIL_ENABLED=true`, an implicit-TLS SMTP endpoint, and +valid `SMTP_HOST`, `SMTP_PORT`, `SMTP_USER`, and `SMTP_PASSWORD` values. Keep +`MAIL_SECURITY_FROM_ADDRESS=security@aiwamediagroup.com` and +`MAIL_ROUTINE_FROM_ADDRESS=creator-tool@aiwamediagroup.com`. + +Application code writes a `MailMessage` outbox row first. The worker dispatches +eligible rows through BullMQ and SMTP. A Redis or SMTP outage therefore does not +discard pending mail. Security mail is high priority, uses idempotency keys, and +has sensitive message bodies redacted after successful delivery. Routine +generation notifications are non-critical to the generation lifecycle: failure +to enqueue or deliver a notification must never turn a successfully stored and +charged generation into a failed generation. + +After deployment, verify migrations were applied and both web and worker +services use the same production environment. Test both sender identities before +enabling customer-facing notification volume. Never expose SMTP credentials +through `NEXT_PUBLIC_*` variables or application logs. diff --git a/packages/config/src/index.ts b/packages/config/src/index.ts index 097a9a344..d10df6f74 100644 --- a/packages/config/src/index.ts +++ b/packages/config/src/index.ts @@ -47,9 +47,28 @@ export const serverEnvSchema = z.object({ APP_URL: z.url().default("http://localhost:3000"), AUTH_SECRET: z.string().min(32), SIGNUPS_ENABLED: booleanFromString, - AUTH_EMAIL_WEBHOOK_URL: optionalHttpsUrl, - AUTH_EMAIL_WEBHOOK_BEARER_TOKEN: optionalString, - AUTH_EMAIL_FROM: optionalString, + MAIL_ENABLED: z + .enum(["true", "false"]) + .default("true") + .transform((value) => value === "true"), + SMTP_HOST: z.string().min(1).default("localhost"), + SMTP_PORT: z.coerce.number().int().min(1).max(65535).default(465), + SMTP_USER: z.string().min(1).default("local"), + SMTP_PASSWORD: z.string().min(1).default("local"), + SMTP_EHLO_NAME: z.string().min(1).default("creator.aiwamediagroup.com"), + SMTP_POOL_MAX_CONNECTIONS: z.coerce.number().int().min(1).max(20).default(3), + SMTP_POOL_MAX_MESSAGES: z.coerce.number().int().min(1).max(1000).default(100), + SMTP_CONNECTION_TIMEOUT_MS: optionalPositiveInteger.default(10_000), + SMTP_GREETING_TIMEOUT_MS: optionalPositiveInteger.default(10_000), + SMTP_SOCKET_TIMEOUT_MS: optionalPositiveInteger.default(30_000), + MAIL_SECURITY_FROM_ADDRESS: z + .string() + .email() + .default("security@aiwamediagroup.com"), + MAIL_ROUTINE_FROM_ADDRESS: z + .string() + .email() + .default("creator-tool@aiwamediagroup.com"), DATABASE_URL: z.string().min(1), REDIS_URL: z.url().default("redis://127.0.0.1:6379/0"), BYTEPLUS_API_KEY: optionalString, @@ -99,5 +118,28 @@ export function parseServerEnv( throw new Error(`Invalid server environment variables: ${fields}`); } + if (result.data.APP_ENV === "production" && result.data.MAIL_ENABLED) { + const requiredMailVariables = [ + "SMTP_HOST", + "SMTP_USER", + "SMTP_PASSWORD", + "MAIL_SECURITY_FROM_ADDRESS", + "MAIL_ROUTINE_FROM_ADDRESS", + ] as const; + const missing = requiredMailVariables.filter( + (key) => !environment[key]?.trim(), + ); + if (missing.length) { + throw new Error( + `Invalid server environment variables: ${missing.join(", ")}`, + ); + } + if (result.data.SMTP_PORT !== 465) { + throw new Error( + "Invalid server environment variables: SMTP_PORT (implicit TLS on port 465 is required in production)", + ); + } + } + return result.data; } diff --git a/packages/config/test/mail-env.test.ts b/packages/config/test/mail-env.test.ts new file mode 100644 index 000000000..cf047c8a8 --- /dev/null +++ b/packages/config/test/mail-env.test.ts @@ -0,0 +1,47 @@ +import { describe, expect, it } from "vitest"; +import { parseServerEnv } from "../src/index"; + +const productionBase = { + NODE_ENV: "production", + APP_ENV: "production", + APP_VERSION: "test", + APP_URL: "https://creator.aiwamediagroup.com", + AUTH_SECRET: "test-auth-secret-that-is-at-least-32-characters", + SIGNUPS_ENABLED: "true", + DATABASE_URL: "mysql://creator:test@127.0.0.1:3306/creator_platform", + REDIS_URL: "redis://127.0.0.1:6379/0", +}; + +describe("production mail configuration", () => { + it("fails closed when mail is enabled without explicit SMTP credentials", () => { + expect(() => + parseServerEnv({ + ...productionBase, + MAIL_ENABLED: "true", + }), + ).toThrow(/SMTP_HOST/); + }); + + it("accepts explicitly disabled mail without SMTP credentials", () => { + const env = parseServerEnv({ + ...productionBase, + MAIL_ENABLED: "false", + }); + expect(env.MAIL_ENABLED).toBe(false); + }); + + it("requires implicit TLS port 465 in production", () => { + expect(() => + parseServerEnv({ + ...productionBase, + MAIL_ENABLED: "true", + SMTP_HOST: "mail.aiwamediagroup.com", + SMTP_PORT: "587", + SMTP_USER: "security@aiwamediagroup.com", + SMTP_PASSWORD: "secret", + MAIL_SECURITY_FROM_ADDRESS: "security@aiwamediagroup.com", + MAIL_ROUTINE_FROM_ADDRESS: "creator-tool@aiwamediagroup.com", + }), + ).toThrow(/SMTP_PORT/); + }); +}); diff --git a/packages/db/prisma/migrations/20260924020000_transactional_mail_outbox/migration.sql b/packages/db/prisma/migrations/20260924020000_transactional_mail_outbox/migration.sql new file mode 100644 index 000000000..9421a7059 --- /dev/null +++ b/packages/db/prisma/migrations/20260924020000_transactional_mail_outbox/migration.sql @@ -0,0 +1,34 @@ +CREATE TABLE `MailMessage` ( + `id` VARCHAR(191) NOT NULL, + `kind` ENUM('SECURITY','ROUTINE') NOT NULL, + `priority` ENUM('HIGH','NORMAL') NOT NULL DEFAULT 'NORMAL', + `status` ENUM('PENDING','QUEUED','SENDING','RETRY','SENT','FAILED','SUPPRESSED') NOT NULL DEFAULT 'PENDING', + `template` VARCHAR(191) NOT NULL, + `recipient` VARCHAR(191) NOT NULL, + `subject` VARCHAR(191) NOT NULL, + `textBody` LONGTEXT NOT NULL, + `htmlBody` LONGTEXT NOT NULL, + `sensitive` BOOLEAN NOT NULL DEFAULT false, + `userId` VARCHAR(191) NULL, + `organizationId` VARCHAR(191) NULL, + `idempotencyKey` VARCHAR(191) NOT NULL, + `attemptCount` INTEGER NOT NULL DEFAULT 0, + `nextAttemptAt` DATETIME(3) NULL, + `providerMessageId` VARCHAR(191) NULL, + `lastErrorCode` VARCHAR(191) NULL, + `lastErrorMessage` TEXT NULL, + `queuedAt` DATETIME(3) NULL, + `sendingAt` DATETIME(3) NULL, + `sentAt` DATETIME(3) NULL, + `failedAt` DATETIME(3) NULL, + `createdAt` DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + `updatedAt` DATETIME(3) NOT NULL, + UNIQUE INDEX `MailMessage_idempotencyKey_key`(`idempotencyKey`), + INDEX `MailMessage_status_priority_createdAt_idx`(`status`, `priority`, `createdAt`), + INDEX `MailMessage_status_nextAttemptAt_idx`(`status`, `nextAttemptAt`), + INDEX `MailMessage_status_sendingAt_idx`(`status`, `sendingAt`), + INDEX `MailMessage_userId_createdAt_idx`(`userId`, `createdAt`), + INDEX `MailMessage_organizationId_createdAt_idx`(`organizationId`, `createdAt`), + INDEX `MailMessage_recipient_createdAt_idx`(`recipient`, `createdAt`), + PRIMARY KEY (`id`) +) DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; diff --git a/packages/db/prisma/migrations/20260924030000_notification_preferences/migration.sql b/packages/db/prisma/migrations/20260924030000_notification_preferences/migration.sql new file mode 100644 index 000000000..e0eeeac38 --- /dev/null +++ b/packages/db/prisma/migrations/20260924030000_notification_preferences/migration.sql @@ -0,0 +1,12 @@ +CREATE TABLE `NotificationPreference` ( + `userId` VARCHAR(191) NOT NULL, + `generationCompleted` BOOLEAN NOT NULL DEFAULT true, + `generationFailed` BOOLEAN NOT NULL DEFAULT true, + `reports` BOOLEAN NOT NULL DEFAULT true, + `createdAt` DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + `updatedAt` DATETIME(3) NOT NULL, + PRIMARY KEY (`userId`), + CONSTRAINT `NotificationPreference_userId_fkey` + FOREIGN KEY (`userId`) REFERENCES `User`(`id`) + ON DELETE CASCADE ON UPDATE CASCADE +) DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; diff --git a/packages/db/prisma/schema.prisma b/packages/db/prisma/schema.prisma index b928f625f..874453cc5 100644 --- a/packages/db/prisma/schema.prisma +++ b/packages/db/prisma/schema.prisma @@ -98,31 +98,32 @@ enum AssetStatus { } model User { - id String @id @default(cuid()) - name String - email String @unique - emailVerified Boolean @default(false) - image String? @db.Text - platformRole PlatformRole @default(USER) - disabledAt DateTime? - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt - sessions Session[] - accounts Account[] - memberships Membership[] - onboardedOrganization Organization? @relation("SelfServeOrganization") - createdPayments ManualPayment[] @relation("PaymentCreatedBy") - confirmedPayments ManualPayment[] @relation("PaymentConfirmedBy") - priceVersions ModelPriceVersion[] @relation("PriceVersionCreatedBy") - generationJobs GenerationJob[] @relation("GenerationCreatedBy") - reasoningJobs ReasoningJob[] @relation("ReasoningCreatedBy") - ownedOrganizations Organization[] @relation("OrganizationOwner") - storageAssets Asset[] @relation("AssetStorageOwner") - auditEvents AuditEvent[] @relation("AuditActor") - createdInvitations OrganizationInvitation[] @relation("InvitationCreatedBy") - acceptedInvitations OrganizationInvitation[] @relation("InvitationAcceptedBy") - twoFactorEnabled Boolean @default(false) - twoFactors TwoFactor[] + id String @id @default(cuid()) + name String + email String @unique + emailVerified Boolean @default(false) + image String? @db.Text + platformRole PlatformRole @default(USER) + disabledAt DateTime? + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + sessions Session[] + accounts Account[] + memberships Membership[] + onboardedOrganization Organization? @relation("SelfServeOrganization") + createdPayments ManualPayment[] @relation("PaymentCreatedBy") + confirmedPayments ManualPayment[] @relation("PaymentConfirmedBy") + priceVersions ModelPriceVersion[] @relation("PriceVersionCreatedBy") + generationJobs GenerationJob[] @relation("GenerationCreatedBy") + reasoningJobs ReasoningJob[] @relation("ReasoningCreatedBy") + ownedOrganizations Organization[] @relation("OrganizationOwner") + storageAssets Asset[] @relation("AssetStorageOwner") + auditEvents AuditEvent[] @relation("AuditActor") + createdInvitations OrganizationInvitation[] @relation("InvitationCreatedBy") + acceptedInvitations OrganizationInvitation[] @relation("InvitationAcceptedBy") + twoFactorEnabled Boolean @default(false) + twoFactors TwoFactor[] + notificationPreference NotificationPreference? @@index([platformRole, disabledAt]) } @@ -192,17 +193,27 @@ model TwoFactor { @@index([userId]) } +model NotificationPreference { + userId String @id + generationCompleted Boolean @default(true) + generationFailed Boolean @default(true) + reports Boolean @default(true) + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + user User @relation(fields: [userId], references: [id], onDelete: Cascade) +} + model Organization { - id String @id @default(cuid()) + id String @id @default(cuid()) name String - slug String @unique - status OrganizationStatus @default(ACTIVE) - selfServeCreatorUserId String? @unique + slug String @unique + status OrganizationStatus @default(ACTIVE) + selfServeCreatorUserId String? @unique ownerUserId String - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt - selfServeCreator User? @relation("SelfServeOrganization", fields: [selfServeCreatorUserId], references: [id], onDelete: SetNull) - owner User @relation("OrganizationOwner", fields: [ownerUserId], references: [id], onDelete: Restrict) + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + selfServeCreator User? @relation("SelfServeOrganization", fields: [selfServeCreatorUserId], references: [id], onDelete: SetNull) + owner User @relation("OrganizationOwner", fields: [ownerUserId], references: [id], onDelete: Restrict) activeSessions Session[] wallet Wallet? memberships Membership[] @@ -220,15 +231,15 @@ model Organization { } model Membership { - id String @id @default(cuid()) - organizationId String - userId String - role MembershipRole @default(ORGANIZATION_MEMBER) + id String @id @default(cuid()) + organizationId String + userId String + role MembershipRole @default(ORGANIZATION_MEMBER) monthlySpendingCapCredits BigInt? - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt - organization Organization @relation(fields: [organizationId], references: [id], onDelete: Cascade) - user User @relation(fields: [userId], references: [id], onDelete: Cascade) + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + organization Organization @relation(fields: [organizationId], references: [id], onDelete: Cascade) + user User @relation(fields: [userId], references: [id], onDelete: Cascade) @@unique([organizationId, userId]) @@index([userId]) @@ -268,33 +279,33 @@ model LedgerEntry { } model ManualPayment { - id String @id @default(cuid()) - organizationId String - method PaymentMethod - status PaymentStatus @default(DRAFT) - amountBaisa BigInt - reference String? - chequeNumber String? - bankName String? - notes String? @db.Text - receivedAt DateTime - confirmedAt DateTime? - rejectedAt DateTime? - reversedAt DateTime? + id String @id @default(cuid()) + organizationId String + method PaymentMethod + status PaymentStatus @default(DRAFT) + amountBaisa BigInt + reference String? + chequeNumber String? + bankName String? + notes String? @db.Text + receivedAt DateTime + confirmedAt DateTime? + rejectedAt DateTime? + reversedAt DateTime? rejectionReason String? @db.Text rejectionIdempotencyKey String? @unique reversalReason String? @db.Text - creditsGranted BigInt? - creditsPerBaisa BigInt? - idempotencyKey String @unique - createdById String - confirmedById String? - ledgerEntryId String? @unique - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt - organization Organization @relation(fields: [organizationId], references: [id], onDelete: Restrict) - createdBy User @relation("PaymentCreatedBy", fields: [createdById], references: [id], onDelete: Restrict) - confirmedBy User? @relation("PaymentConfirmedBy", fields: [confirmedById], references: [id], onDelete: Restrict) + creditsGranted BigInt? + creditsPerBaisa BigInt? + idempotencyKey String @unique + createdById String + confirmedById String? + ledgerEntryId String? @unique + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + organization Organization @relation(fields: [organizationId], references: [id], onDelete: Restrict) + createdBy User @relation("PaymentCreatedBy", fields: [createdById], references: [id], onDelete: Restrict) + confirmedBy User? @relation("PaymentConfirmedBy", fields: [confirmedById], references: [id], onDelete: Restrict) @@index([organizationId, status]) @@index([method, receivedAt]) @@ -398,25 +409,25 @@ model GenerationJob { } model Asset { - id String @id @default(cuid()) - organizationId String - projectId String? - generationJobId String? + id String @id @default(cuid()) + organizationId String + projectId String? + generationJobId String? storageOwnerUserId String? - status AssetStatus @default(PENDING) - objectKey String @unique - mimeType String - byteSize BigInt - sha256 String? - width Int? - height Int? - durationMs Int? - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt - organization Organization @relation(fields: [organizationId], references: [id], onDelete: Restrict) - project Project? @relation(fields: [projectId], references: [id], onDelete: SetNull) - generationJob GenerationJob? @relation(fields: [generationJobId], references: [id], onDelete: SetNull) - storageOwner User? @relation("AssetStorageOwner", fields: [storageOwnerUserId], references: [id], onDelete: SetNull) + status AssetStatus @default(PENDING) + objectKey String @unique + mimeType String + byteSize BigInt + sha256 String? + width Int? + height Int? + durationMs Int? + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + organization Organization @relation(fields: [organizationId], references: [id], onDelete: Restrict) + project Project? @relation(fields: [projectId], references: [id], onDelete: SetNull) + generationJob GenerationJob? @relation(fields: [generationJobId], references: [id], onDelete: SetNull) + storageOwner User? @relation("AssetStorageOwner", fields: [storageOwnerUserId], references: [id], onDelete: SetNull) @@index([organizationId, createdAt]) @@index([generationJobId]) @@ -466,6 +477,60 @@ model OrganizationInvitation { @@index([email]) } +enum MailKind { + SECURITY + ROUTINE +} + +enum MailPriority { + HIGH + NORMAL +} + +enum MailStatus { + PENDING + QUEUED + SENDING + RETRY + SENT + FAILED + SUPPRESSED +} + +model MailMessage { + id String @id @default(cuid()) + kind MailKind + priority MailPriority @default(NORMAL) + status MailStatus @default(PENDING) + template String + recipient String + subject String + textBody String @db.LongText + htmlBody String @db.LongText + sensitive Boolean @default(false) + userId String? + organizationId String? + idempotencyKey String @unique + attemptCount Int @default(0) + nextAttemptAt DateTime? + providerMessageId String? + lastErrorCode String? + lastErrorMessage String? @db.Text + queuedAt DateTime? + sendingAt DateTime? + sentAt DateTime? + failedAt DateTime? + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + + @@index([status, priority, createdAt]) + @@index([status, nextAttemptAt]) + @@index([status, sendingAt]) + @@index([userId, createdAt]) + @@index([organizationId, createdAt]) + @@index([recipient, createdAt]) +} + enum ReasoningJobStatus { QUEUED PROCESSING @@ -474,28 +539,28 @@ enum ReasoningJobStatus { } model ReasoningJob { - id String @id @default(cuid()) - organizationId String - createdById String - providerModelId String - status ReasoningJobStatus @default(QUEUED) - idempotencyKey String @unique - requestPayload Json - outputPayload Json? - providerRequestId String? - inputTokens Int? - outputTokens Int? - errorCode String? - errorMessage String? @db.Text - queuedAt DateTime? - processingAt DateTime? - completedAt DateTime? - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt - - organization Organization @relation(fields: [organizationId], references: [id], onDelete: Restrict) - createdBy User @relation("ReasoningCreatedBy", fields: [createdById], references: [id], onDelete: Restrict) - providerModel ProviderModel @relation(fields: [providerModelId], references: [id], onDelete: Restrict) + id String @id @default(cuid()) + organizationId String + createdById String + providerModelId String + status ReasoningJobStatus @default(QUEUED) + idempotencyKey String @unique + requestPayload Json + outputPayload Json? + providerRequestId String? + inputTokens Int? + outputTokens Int? + errorCode String? + errorMessage String? @db.Text + queuedAt DateTime? + processingAt DateTime? + completedAt DateTime? + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + + organization Organization @relation(fields: [organizationId], references: [id], onDelete: Restrict) + createdBy User @relation("ReasoningCreatedBy", fields: [createdById], references: [id], onDelete: Restrict) + providerModel ProviderModel @relation(fields: [providerModelId], references: [id], onDelete: Restrict) @@index([organizationId, status, createdAt]) @@index([createdById, organizationId, status, createdAt]) @@ -503,4 +568,3 @@ model ReasoningJob { @@index([status, updatedAt]) @@index([providerRequestId]) } - diff --git a/packages/generation/package.json b/packages/generation/package.json index dc5cf466a..b38833117 100644 --- a/packages/generation/package.json +++ b/packages/generation/package.json @@ -19,6 +19,7 @@ "dependencies": { "@aiwa/credits": "workspace:*", "@aiwa/db": "workspace:*", + "@aiwa/mail": "workspace:*", "@aiwa/authz": "workspace:*", "@aiwa/config": "workspace:*", "@aiwa/organizations": "workspace:*", diff --git a/packages/generation/src/process.ts b/packages/generation/src/process.ts index 8503d1cae..bedaf8665 100644 --- a/packages/generation/src/process.ts +++ b/packages/generation/src/process.ts @@ -1,5 +1,10 @@ import { captureCreditsForJob, releaseOrRefundCredits } from "@aiwa/credits"; -import { db } from "@aiwa/db"; +import { db, type Prisma } from "@aiwa/db"; +import { + enqueueMail, + generationCompletedEmail, + generationFailedEmail, +} from "@aiwa/mail"; import { ProviderConfigurationError, ProviderRequestError, @@ -16,6 +21,82 @@ import { storedAssetSize, } from "./storage"; +async function generationRecipient( + tx: Prisma.TransactionClient, + userId: string, + category: "generationCompleted" | "generationFailed", +): Promise { + const runtimeTx = tx as Prisma.TransactionClient & { + user?: Prisma.TransactionClient["user"]; + mailMessage?: Prisma.TransactionClient["mailMessage"]; + notificationPreference?: Prisma.TransactionClient["notificationPreference"]; + }; + if (!runtimeTx.user || !runtimeTx.mailMessage) return null; + const user = await runtimeTx.user.findUnique({ + where: { id: userId }, + select: { + email: true, + disabledAt: true, + notificationPreference: { + select: { + generationCompleted: true, + generationFailed: true, + }, + }, + }, + }); + if (!user || user.disabledAt) return null; + const preferences = user.notificationPreference; + if (preferences && preferences[category] === false) return null; + return user.email; +} + +async function enqueueGenerationFailure( + tx: Prisma.TransactionClient, + job: { id: string; organizationId: string; createdById: string }, + message: string, +): Promise { + const to = await generationRecipient(tx, job.createdById, "generationFailed"); + if (!to) return; + await enqueueMail( + generationFailedEmail({ + to, + userId: job.createdById, + organizationId: job.organizationId, + generationJobId: job.id, + message, + }), + tx, + ); +} + +async function enqueueGenerationSuccess( + tx: Prisma.TransactionClient, + job: { id: string; organizationId: string; createdById: string }, + assetId: string, +): Promise { + const to = await generationRecipient( + tx, + job.createdById, + "generationCompleted", + ); + if (!to) return; + const appUrl = process.env.APP_URL ?? "http://localhost:3000"; + await enqueueMail( + generationCompletedEmail({ + to, + userId: job.createdById, + organizationId: job.organizationId, + generationJobId: job.id, + assetUrl: new URL( + `/api/assets/${encodeURIComponent(assetId)}`, + appUrl, + ).toString(), + }), + tx, + ); +} + export async function failJob( id: string, message: string, @@ -50,6 +131,7 @@ export async function failJob( completedAt: new Date(), }, }); + await enqueueGenerationFailure(tx, job, message); }); } @@ -240,7 +322,7 @@ export async function processVideoPollJob( amountCredits: current.reservedCredits, idempotencyKey: `generation-capture-${id}`, }); - await tx.asset.update({ + const asset = await tx.asset.update({ where: { objectKey: `${id}.mp4` }, data: { ...stored, status: "READY" }, }); @@ -263,6 +345,9 @@ export async function processVideoPollJob( targetId: id, }, }); + if (asset?.id) { + await enqueueGenerationSuccess(tx, { ...job, id }, asset.id); + } }); } @@ -386,7 +471,7 @@ export async function processImageJob( amountCredits: current.reservedCredits, idempotencyKey: `generation-capture-${id}`, }); - await tx.asset.update({ + const asset = await tx.asset.update({ where: { objectKey: `${id}.png` }, data: { ...stored, status: "READY" }, }); @@ -409,6 +494,9 @@ export async function processImageJob( targetId: id, }, }); + if (asset?.id) { + await enqueueGenerationSuccess(tx, { ...job, id }, asset.id); + } }); } @@ -639,7 +727,7 @@ async function finalizeVoiceJob( amountCredits: current.reservedCredits, idempotencyKey: `generation-capture-${id}`, }); - await tx.asset.update({ + const asset = await tx.asset.update({ where: { objectKey: `${id}.mp3` }, data: { ...stored, status: "READY" }, }); @@ -672,5 +760,8 @@ async function finalizeVoiceJob( targetId: id, }, }); + if (asset?.id) { + await enqueueGenerationSuccess(tx, { ...job, id }, asset.id); + } }); } diff --git a/packages/mail/package.json b/packages/mail/package.json new file mode 100644 index 000000000..1778292e7 --- /dev/null +++ b/packages/mail/package.json @@ -0,0 +1,26 @@ +{ + "name": "@aiwa/mail", + "version": "0.1.0", + "private": true, + "type": "module", + "exports": { + ".": "./src/index.ts", + "./transport": "./src/transport.ts" + }, + "scripts": { + "build": "tsc --noEmit", + "clean": "rm -rf dist .turbo", + "lint": "eslint src test --max-warnings=0", + "test": "vitest run", + "typecheck": "tsc --noEmit" + }, + "dependencies": { + "@aiwa/config": "workspace:*", + "@aiwa/db": "workspace:*", + "nodemailer": "9.1.1" + }, + "devDependencies": { + "@types/nodemailer": "8.0.2", + "vitest": "^4.1.11" + } +} diff --git a/packages/mail/src/index.ts b/packages/mail/src/index.ts new file mode 100644 index 000000000..3a9a33b02 --- /dev/null +++ b/packages/mail/src/index.ts @@ -0,0 +1,432 @@ +import { db, type Prisma } from "@aiwa/db"; +import type { ServerEnv } from "@aiwa/config"; + +export type MailKind = "SECURITY" | "ROUTINE"; +export type MailPriority = "HIGH" | "NORMAL"; + +export type MailDraft = { + kind: MailKind; + template: string; + to: string; + subject: string; + text: string; + html: string; + sensitive?: boolean; + organizationId?: string | null; + userId?: string | null; + priority?: MailPriority; + idempotencyKey: string; +}; + +type MailDb = typeof db | Prisma.TransactionClient; + +export function senderForKind(env: ServerEnv, kind: MailKind): string { + return kind === "SECURITY" + ? env.MAIL_SECURITY_FROM_ADDRESS + : env.MAIL_ROUTINE_FROM_ADDRESS; +} + +export function mailJobId(id: string): string { + return `mail:${id}`; +} + +export function normalizeRecipient(value: string): string { + return value.trim().toLowerCase(); +} + +export async function enqueueMail( + input: MailDraft, + client: MailDb = db, +): Promise<{ id: string; created: boolean }> { + const to = normalizeRecipient(input.to); + try { + const row = await client.mailMessage.create({ + data: { + kind: input.kind, + priority: + input.priority ?? (input.kind === "SECURITY" ? "HIGH" : "NORMAL"), + template: input.template, + recipient: to, + subject: input.subject, + textBody: input.text, + htmlBody: input.html, + sensitive: input.sensitive ?? input.kind === "SECURITY", + organizationId: input.organizationId ?? null, + userId: input.userId ?? null, + idempotencyKey: input.idempotencyKey, + status: "PENDING", + }, + select: { id: true }, + }); + return { id: row.id, created: true }; + } catch (error) { + if ( + typeof error === "object" && + error !== null && + "code" in error && + error.code === "P2002" + ) { + const existing = await client.mailMessage.findUnique({ + where: { idempotencyKey: input.idempotencyKey }, + select: { id: true }, + }); + if (existing) return { id: existing.id, created: false }; + } + throw error; + } +} + +function escapeHtml(value: string): string { + return value.replace( + /[&<>"']/g, + (char) => + ({ + "&": "&", + "<": "<", + ">": ">", + '"': """, + "'": "'", + })[char] ?? char, + ); +} + +function emailShell(title: string, body: string): string { + return `
Aiwa Creators

${escapeHtml(title)}

${body}

This is an automated message from Aiwa Creators.

`; +} + +export function verificationEmail(input: { + to: string; + verificationUrl: string; + idempotencyKey: string; + userId?: string; +}): MailDraft { + const safeUrl = escapeHtml(input.verificationUrl); + return { + kind: "SECURITY", + template: "auth.verify_email.v1", + to: input.to, + subject: "Verify your Aiwa Creators email", + text: `Verify your email address to activate workspace access: ${input.verificationUrl}\n\nIf you did not request this, you can ignore this message.`, + html: emailShell( + "Verify your email", + `

Confirm this email address to activate workspace access.

Verify email

If the button does not work, copy this link:
${safeUrl}

`, + ), + sensitive: true, + userId: input.userId, + idempotencyKey: input.idempotencyKey, + }; +} + +export function passwordResetEmail(input: { + to: string; + resetUrl: string; + idempotencyKey: string; + userId?: string; +}): MailDraft { + const safeUrl = escapeHtml(input.resetUrl); + return { + kind: "SECURITY", + template: "auth.password_reset.v1", + to: input.to, + subject: "Reset your Aiwa Creators password", + text: `Use this link to reset your Aiwa Creators password: ${input.resetUrl}\n\nIf you did not request this, ignore this email and your password will remain unchanged.`, + html: emailShell( + "Reset your password", + `

A password reset was requested for your Aiwa Creators account.

Reset password

If the button does not work, copy this link:
${safeUrl}

If you did not request this, you can safely ignore this message.

`, + ), + sensitive: true, + userId: input.userId, + idempotencyKey: input.idempotencyKey, + }; +} + +export function securityEventEmail(input: { + to: string; + userId: string; + event: + | "PASSWORD_RESET" + | "MFA_ENABLED" + | "MFA_DISABLED" + | "BACKUP_CODES_REGENERATED"; + idempotencyKey: string; +}): MailDraft { + const content = { + PASSWORD_RESET: { + subject: "Your Aiwa Creators password was changed", + title: "Password changed", + message: + "Your account password was successfully changed and existing sessions were revoked. If this was not you, contact your administrator immediately.", + }, + MFA_ENABLED: { + subject: "Two-factor authentication enabled", + title: "Two-factor authentication enabled", + message: + "Authenticator-based two-factor authentication was enabled on your Aiwa Creators account.", + }, + MFA_DISABLED: { + subject: "Two-factor authentication disabled", + title: "Two-factor authentication disabled", + message: + "Two-factor authentication was disabled on your Aiwa Creators account. If this was not you, contact your administrator immediately.", + }, + BACKUP_CODES_REGENERATED: { + subject: "New two-factor recovery codes generated", + title: "Recovery codes regenerated", + message: + "New two-factor recovery codes were generated for your Aiwa Creators account. Previous recovery codes are no longer valid.", + }, + }[input.event]; + + return { + kind: "SECURITY", + template: `auth.security_event.${input.event.toLowerCase()}.v1`, + to: input.to, + subject: content.subject, + text: content.message, + html: emailShell( + content.title, + `

${escapeHtml(content.message)}

`, + ), + sensitive: false, + userId: input.userId, + idempotencyKey: input.idempotencyKey, + }; +} + +export function accountAdministrationEmail(input: { + to: string; + userId: string; + event: + | "PLATFORM_ROLE_CHANGED" + | "ACCOUNT_DISABLED" + | "ACCOUNT_ENABLED" + | "SESSIONS_REVOKED"; + detail: string; + eventVersion: string; +}): MailDraft { + const content = { + PLATFORM_ROLE_CHANGED: { + subject: "Your Aiwa Creators platform role changed", + title: "Platform role changed", + }, + ACCOUNT_DISABLED: { + subject: "Your Aiwa Creators account was disabled", + title: "Account disabled", + }, + ACCOUNT_ENABLED: { + subject: "Your Aiwa Creators account was reactivated", + title: "Account reactivated", + }, + SESSIONS_REVOKED: { + subject: "Aiwa Creators sessions were revoked", + title: "Sessions revoked", + }, + }[input.event]; + + return { + kind: "SECURITY", + template: `account.admin_${input.event.toLowerCase()}.v1`, + to: input.to, + subject: content.subject, + text: input.detail, + html: emailShell( + content.title, + `

${escapeHtml(input.detail)}

`, + ), + userId: input.userId, + sensitive: false, + idempotencyKey: `account:${input.userId}:${input.event.toLowerCase()}:${input.eventVersion}`, + }; +} + +export function teamMemberAddedEmail(input: { + to: string; + organizationName: string; + role: string; + organizationId: string; + userId: string; + membershipId: string; +}): MailDraft { + return { + kind: "SECURITY", + template: "organization.member_added.v1", + to: input.to, + subject: `You were added to ${input.organizationName} on Aiwa Creators`, + text: `You were added to ${input.organizationName} with role ${input.role}. Sign in to Aiwa Creators to access the workspace.`, + html: emailShell( + "You joined a workspace", + `

You were added to ${escapeHtml(input.organizationName)} with role ${escapeHtml(input.role)}.

Sign in to Aiwa Creators to access the workspace.

`, + ), + organizationId: input.organizationId, + userId: input.userId, + idempotencyKey: `membership-added:${input.membershipId}`, + }; +} + +export function teamMembershipChangedEmail(input: { + to: string; + organizationName: string; + organizationId: string; + userId: string; + membershipId: string; + event: + | "ROLE_CHANGED" + | "CAP_CHANGED" + | "REMOVED" + | "OWNER_GRANTED" + | "OWNER_RELEASED"; + detail?: string; + eventVersion: string; +}): MailDraft { + const content = { + ROLE_CHANGED: { + subject: `Your role changed in ${input.organizationName}`, + title: "Workspace role changed", + message: input.detail ?? "Your workspace role was changed.", + }, + CAP_CHANGED: { + subject: `Your spending limit changed in ${input.organizationName}`, + title: "Spending limit changed", + message: input.detail ?? "Your workspace spending limit was changed.", + }, + REMOVED: { + subject: `Your access to ${input.organizationName} was removed`, + title: "Workspace access removed", + message: "Your membership in this workspace was removed.", + }, + OWNER_GRANTED: { + subject: `You are now the owner of ${input.organizationName}`, + title: "Workspace ownership transferred to you", + message: "You are now the organization owner for this workspace.", + }, + OWNER_RELEASED: { + subject: `Ownership changed for ${input.organizationName}`, + title: "Workspace ownership transferred", + message: + "You are no longer the organization owner. Your workspace membership remains active.", + }, + }[input.event]; + + return { + kind: "SECURITY", + template: `organization.membership_${input.event.toLowerCase()}.v1`, + to: input.to, + subject: content.subject, + text: `${content.message} Organization: ${input.organizationName}.`, + html: emailShell( + content.title, + `

${escapeHtml(content.message)}

Workspace: ${escapeHtml(input.organizationName)}

`, + ), + organizationId: input.organizationId, + userId: input.userId, + sensitive: false, + idempotencyKey: `membership:${input.membershipId}:${input.event.toLowerCase()}:${input.eventVersion}`, + }; +} + +export function billingStatusEmail(input: { + to: string; + organizationName: string; + organizationId: string; + paymentId: string; + status: "CONFIRMED" | "REJECTED" | "REVERSED"; + amountBaisa: bigint; + userId: string; + detail?: string; +}): MailDraft { + const amountOmr = (Number(input.amountBaisa) / 1000).toFixed(3); + const action = + input.status === "CONFIRMED" + ? "confirmed" + : input.status === "REJECTED" + ? "rejected" + : "reversed"; + return { + kind: "SECURITY", + template: `billing.payment_${action}.v1`, + to: input.to, + subject: `Payment ${action} for ${input.organizationName}`, + text: `A payment of OMR ${amountOmr} for ${input.organizationName} was ${action}.${input.detail ? ` ${input.detail}` : ""}`, + html: emailShell( + `Payment ${action}`, + `

A payment of OMR ${amountOmr} for ${escapeHtml(input.organizationName)} was ${action}.

${input.detail ? `

${escapeHtml(input.detail)}

` : ""}`, + ), + organizationId: input.organizationId, + userId: input.userId, + idempotencyKey: `billing:${input.paymentId}:${input.status.toLowerCase()}`, + }; +} + +export function invitationEmail(input: { + to: string; + organizationName: string; + organizationId: string; + role: string; + invitationId: string; + invitationUrl: string; +}): MailDraft { + const safeUrl = escapeHtml(input.invitationUrl); + return { + kind: "SECURITY", + template: "organization.invitation.v1", + to: input.to, + subject: `You're invited to ${input.organizationName} on Aiwa Creators`, + text: `You were invited to ${input.organizationName} with role ${input.role}. Accept the invitation: ${input.invitationUrl}`, + html: emailShell( + "Workspace invitation", + `

You were invited to ${escapeHtml(input.organizationName)} with role ${escapeHtml(input.role)}.

Accept invitation

`, + ), + organizationId: input.organizationId, + sensitive: true, + idempotencyKey: `invitation:${input.invitationId}`, + }; +} + +export function generationFailedEmail(input: { + to: string; + userId: string; + organizationId: string; + generationJobId: string; + message: string; +}): MailDraft { + return { + kind: "ROUTINE", + template: "generation.failed.v1", + to: input.to, + subject: "Your Aiwa Creators generation could not be completed", + text: `Your generation could not be completed. ${input.message}`, + html: emailShell( + "Generation not completed", + `

Your generation could not be completed.

${escapeHtml(input.message)}

`, + ), + organizationId: input.organizationId, + userId: input.userId, + sensitive: false, + idempotencyKey: `generation-failed:${input.generationJobId}`, + }; +} + +export function generationCompletedEmail(input: { + to: string; + assetUrl: string; + userId: string; + organizationId: string; + generationJobId: string; +}): MailDraft { + const safeUrl = escapeHtml(input.assetUrl); + return { + kind: "ROUTINE", + template: "generation.completed.v1", + to: input.to, + subject: "Your Aiwa Creators generation is ready", + text: `Your generation is ready: ${input.assetUrl}`, + html: emailShell( + "Your generation is ready", + `

Your media has finished generating.

Open generation

`, + ), + organizationId: input.organizationId, + userId: input.userId, + sensitive: false, + idempotencyKey: `generation-completed:${input.generationJobId}`, + }; +} diff --git a/packages/mail/src/transport.ts b/packages/mail/src/transport.ts new file mode 100644 index 000000000..1f0bd107b --- /dev/null +++ b/packages/mail/src/transport.ts @@ -0,0 +1,223 @@ +import type { ServerEnv } from "@aiwa/config"; +import { db } from "@aiwa/db"; +import nodemailer, { type Transporter } from "nodemailer"; +import type SMTPPool from "nodemailer/lib/smtp-pool"; +import { senderForKind } from "./index"; + +let transporter: Transporter< + SMTPPool.SentMessageInfo, + SMTPPool.Options +> | null = null; +let transporterFingerprint = ""; + +function getTransporter( + env: ServerEnv, +): Transporter { + const fingerprint = [ + env.SMTP_HOST, + env.SMTP_PORT, + env.SMTP_USER, + env.SMTP_POOL_MAX_CONNECTIONS, + env.SMTP_POOL_MAX_MESSAGES, + ].join(":"); + + if (transporter && transporterFingerprint === fingerprint) return transporter; + + transporter?.close(); + transporterFingerprint = fingerprint; + const nextTransporter = nodemailer.createTransport({ + host: env.SMTP_HOST, + port: env.SMTP_PORT, + secure: true, + pool: true, + maxConnections: env.SMTP_POOL_MAX_CONNECTIONS, + maxMessages: env.SMTP_POOL_MAX_MESSAGES, + connectionTimeout: env.SMTP_CONNECTION_TIMEOUT_MS, + greetingTimeout: env.SMTP_GREETING_TIMEOUT_MS, + socketTimeout: env.SMTP_SOCKET_TIMEOUT_MS, + auth: { + user: env.SMTP_USER, + pass: env.SMTP_PASSWORD, + }, + tls: { + servername: env.SMTP_HOST, + rejectUnauthorized: true, + }, + }); + + transporter = nextTransporter; + return nextTransporter; +} + +export async function verifySmtpTransport(env: ServerEnv): Promise { + await getTransporter(env).verify(); +} + +export async function closeSmtpTransport(): Promise { + transporter?.close(); + transporter = null; + transporterFingerprint = ""; +} + +export async function sendMailViaSmtp( + env: ServerEnv, + message: { + from: string; + to: string; + subject: string; + text: string; + html: string; + messageId?: string; + }, +): Promise<{ providerMessageId: string }> { + const info = await getTransporter(env).sendMail({ + from: { name: "Aiwa Creators", address: message.from }, + to: message.to, + subject: message.subject, + text: message.text, + html: message.html, + messageId: message.messageId, + }); + + return { providerMessageId: info.messageId }; +} + +export async function processMailMessage( + env: ServerEnv, + id: string, + attemptNumber: number, + maxAttempts: number, +): Promise<{ sent: boolean; terminal: boolean }> { + const current = await db.mailMessage.findUnique({ where: { id } }); + if (!current) return { sent: false, terminal: true }; + if (["SENT", "FAILED", "SUPPRESSED"].includes(current.status)) { + return { sent: current.status === "SENT", terminal: true }; + } + + const claimed = await db.mailMessage.updateMany({ + where: { id, status: { in: ["PENDING", "QUEUED", "RETRY"] } }, + data: { + status: "SENDING", + sendingAt: new Date(), + attemptCount: { increment: 1 }, + lastErrorCode: null, + lastErrorMessage: null, + }, + }); + if (claimed.count !== 1) return { sent: false, terminal: false }; + + const message = await db.mailMessage.findUniqueOrThrow({ where: { id } }); + try { + const result = await sendMailViaSmtp(env, { + from: senderForKind(env, message.kind), + to: message.recipient, + subject: message.subject, + text: message.textBody, + html: message.htmlBody, + messageId: ``, + }); + await db.mailMessage.update({ + where: { id }, + data: { + status: "SENT", + providerMessageId: result.providerMessageId, + sentAt: new Date(), + nextAttemptAt: null, + sendingAt: null, + textBody: message.sensitive + ? "[redacted after delivery]" + : message.textBody, + htmlBody: message.sensitive + ? "[redacted after delivery]" + : message.htmlBody, + }, + }); + return { sent: true, terminal: true }; + } catch (error) { + const failure = classifySmtpFailure(error); + const terminal = !failure.retryable || attemptNumber >= maxAttempts; + const highDelays = [30_000, 120_000, 600_000, 1_800_000]; + const normalDelays = [60_000, 300_000, 1_200_000, 3_600_000]; + const delays = message.priority === "HIGH" ? highDelays : normalDelays; + const delay = + delays[Math.min(Math.max(attemptNumber - 1, 0), delays.length - 1)] ?? + delays[delays.length - 1]!; + await db.mailMessage.update({ + where: { id }, + data: { + status: terminal ? "FAILED" : "RETRY", + failedAt: terminal ? new Date() : null, + nextAttemptAt: terminal ? null : new Date(Date.now() + delay), + sendingAt: null, + lastErrorCode: failure.code, + lastErrorMessage: failure.message.slice(0, 2000), + }, + }); + if (!terminal) throw error; + return { sent: false, terminal: true }; + } +} + +export async function recoverStaleMailDeliveries( + staleBefore: Date, +): Promise { + const result = await db.mailMessage.updateMany({ + where: { + status: "SENDING", + sendingAt: { lt: staleBefore }, + }, + data: { + status: "RETRY", + sendingAt: null, + nextAttemptAt: new Date(), + lastErrorCode: "WORKER_INTERRUPTED", + lastErrorMessage: + "Mail delivery was interrupted before completion was recorded. Delivery will be retried with the same Message-ID.", + }, + }); + return result.count; +} + +export function classifySmtpFailure(error: unknown): { + retryable: boolean; + code: string; + message: string; +} { + const message = error instanceof Error ? error.message : "Unknown SMTP error"; + const responseCode = + typeof error === "object" && + error !== null && + "responseCode" in error && + typeof error.responseCode === "number" + ? error.responseCode + : undefined; + const transportCode = + typeof error === "object" && + error !== null && + "code" in error && + typeof error.code === "string" + ? error.code + : undefined; + + if (responseCode && responseCode >= 500) { + return { retryable: false, code: `SMTP_${responseCode}`, message }; + } + + if (transportCode === "EENVELOPE" && responseCode && responseCode >= 400) { + return { + retryable: responseCode < 500, + code: `SMTP_${responseCode}`, + message, + }; + } + + return { + retryable: true, + code: responseCode + ? `SMTP_${responseCode}` + : transportCode + ? `SMTP_${transportCode}` + : "SMTP_TRANSPORT", + message, + }; +} diff --git a/packages/mail/test/mail.test.ts b/packages/mail/test/mail.test.ts new file mode 100644 index 000000000..57a9ac5de --- /dev/null +++ b/packages/mail/test/mail.test.ts @@ -0,0 +1,50 @@ +import { describe, expect, it } from "vitest"; +import { generationCompletedEmail, senderForKind } from "../src/index"; +import { classifySmtpFailure } from "../src/transport"; + +describe("mail helpers", () => { + it("keeps sender identities separated", () => { + const env = { + MAIL_SECURITY_FROM_ADDRESS: "security@aiwamediagroup.com", + MAIL_ROUTINE_FROM_ADDRESS: "creator-tool@aiwamediagroup.com", + } as never; + expect(senderForKind(env, "SECURITY")).toBe("security@aiwamediagroup.com"); + expect(senderForKind(env, "ROUTINE")).toBe( + "creator-tool@aiwamediagroup.com", + ); + }); + + it("builds idempotent routine generation mail", () => { + const draft = generationCompletedEmail({ + to: "USER@EXAMPLE.COM ", + assetUrl: "https://creator.aiwamediagroup.com/assets/a", + userId: "u1", + organizationId: "o1", + generationJobId: "j1", + }); + expect(draft.kind).toBe("ROUTINE"); + expect(draft.idempotencyKey).toBe("generation-completed:j1"); + }); + + it("treats 5xx SMTP responses as permanent", () => { + const error = Object.assign(new Error("rejected"), { + code: "EENVELOPE", + responseCode: 550, + }); + expect(classifySmtpFailure(error)).toMatchObject({ + retryable: false, + code: "SMTP_550", + }); + }); + + it("treats 4xx SMTP responses as retryable", () => { + const error = Object.assign(new Error("mailbox temporarily unavailable"), { + code: "EENVELOPE", + responseCode: 451, + }); + expect(classifySmtpFailure(error)).toMatchObject({ + retryable: true, + code: "SMTP_451", + }); + }); +}); diff --git a/packages/mail/tsconfig.json b/packages/mail/tsconfig.json new file mode 100644 index 000000000..e89e187c8 --- /dev/null +++ b/packages/mail/tsconfig.json @@ -0,0 +1,5 @@ +{ + "extends": "../../tsconfig.base.json", + "compilerOptions": { "outDir": "dist", "rootDir": "." }, + "include": ["src/**/*.ts", "test/**/*.ts"] +} diff --git a/packages/organizations/package.json b/packages/organizations/package.json index d4ff86aff..94f24088d 100644 --- a/packages/organizations/package.json +++ b/packages/organizations/package.json @@ -15,7 +15,8 @@ }, "dependencies": { "@aiwa/authz": "workspace:*", - "@aiwa/db": "workspace:*" + "@aiwa/db": "workspace:*", + "@aiwa/mail": "workspace:*" }, "devDependencies": { "vitest": "^4.1.11" diff --git a/packages/organizations/src/index.ts b/packages/organizations/src/index.ts index aac779e6f..1b1ef8720 100644 --- a/packages/organizations/src/index.ts +++ b/packages/organizations/src/index.ts @@ -6,6 +6,13 @@ import { type PlatformRole, } from "@aiwa/authz"; import { db, Prisma } from "@aiwa/db"; +import { + accountAdministrationEmail, + enqueueMail, + invitationEmail, + teamMemberAddedEmail, + teamMembershipChangedEmail, +} from "@aiwa/mail"; export const MAX_ORGANIZATION_NON_OWNER_MEMBERS = 9; export const MAX_ORGANIZATION_SEATS = 10; @@ -265,6 +272,10 @@ function authorize( return; throw new PermissionDeniedError(); } +function entityVersion(updatedAt: unknown, fallback: string): string { + return updatedAt instanceof Date ? updatedAt.getTime().toString() : fallback; +} + async function audit( tx: Prisma.TransactionClient, actor: OrganizationActor, @@ -293,7 +304,7 @@ async function lockedOrganization( await tx.$queryRaw`SELECT id FROM Organization WHERE id = ${organizationId} FOR UPDATE`; const organization = await tx.organization.findUnique({ where: { id: organizationId }, - select: { id: true, status: true, ownerUserId: true }, + select: { id: true, status: true, ownerUserId: true, name: true }, }); if (!organization) throw new OrganizationNotFoundError(); if (!allowSuspended && organization.status !== "ACTIVE") @@ -312,16 +323,26 @@ export async function addMember(input: { authorize(input.actor, input.organizationId, "members"); return db.$transaction( async (tx) => { - await lockedOrganization(tx, input.organizationId); + const organization = await lockedOrganization(tx, input.organizationId); const user = input.userId ? await tx.user.findUnique({ where: { id: input.userId }, - select: { id: true, emailVerified: true, disabledAt: true }, + select: { + id: true, + email: true, + emailVerified: true, + disabledAt: true, + }, }) : input.email ? await tx.user.findUnique({ where: { email: normalizeMemberEmail(input.email) }, - select: { id: true, emailVerified: true, disabledAt: true }, + select: { + id: true, + email: true, + emailVerified: true, + disabledAt: true, + }, }) : null; if (!user || user.disabledAt) throw new UserUnavailableError(); @@ -366,6 +387,17 @@ export async function addMember(input: { member.id, { membershipId: member.id, userId: user.id }, ); + await enqueueMail( + teamMemberAddedEmail({ + to: user.email, + organizationName: organization.name, + role: member.role, + organizationId: input.organizationId, + userId: user.id, + membershipId: member.id, + }), + tx, + ); return member; }, { isolationLevel: Prisma.TransactionIsolationLevel.Serializable }, @@ -384,6 +416,7 @@ export async function updateMember(input: { const organization = await lockedOrganization(tx, input.organizationId); const member = await tx.membership.findFirst({ where: { id: input.membershipId, organizationId: input.organizationId }, + include: { user: { select: { email: true } } }, }); if (!member) throw new OrganizationNotFoundError(); if ( @@ -413,6 +446,26 @@ export async function updateMember(input: { member.id, { membershipId: member.id, userId: member.userId }, ); + await enqueueMail( + teamMembershipChangedEmail({ + to: member.user.email, + organizationName: organization.name, + organizationId: input.organizationId, + userId: member.userId, + membershipId: member.id, + event: input.role ? "ROLE_CHANGED" : "CAP_CHANGED", + detail: input.role + ? `Your workspace role is now ${updated.role}.` + : updated.monthlySpendingCapCredits === null + ? "Your monthly spending cap was removed." + : `Your monthly spending cap is now ${updated.monthlySpendingCapCredits.toString()} credits.`, + eventVersion: entityVersion( + updated.updatedAt, + `member:${updated.role}:${updated.monthlySpendingCapCredits?.toString() ?? "none"}`, + ), + }), + tx, + ); return updated; }); } @@ -427,6 +480,7 @@ export async function removeMember(input: { const organization = await lockedOrganization(tx, input.organizationId); const member = await tx.membership.findFirst({ where: { id: input.membershipId, organizationId: input.organizationId }, + include: { user: { select: { email: true } } }, }); if (!member) throw new OrganizationNotFoundError(); if ( @@ -451,6 +505,18 @@ export async function removeMember(input: { member.id, { membershipId: member.id, userId: member.userId }, ); + await enqueueMail( + teamMembershipChangedEmail({ + to: member.user.email, + organizationName: organization.name, + organizationId: input.organizationId, + userId: member.userId, + membershipId: member.id, + event: "REMOVED", + eventVersion: "removed", + }), + tx, + ); }); } @@ -468,6 +534,7 @@ export async function transferOwnership(input: { organizationId: input.organizationId, user: { disabledAt: null }, }, + include: { user: { select: { email: true } } }, }); if (!target || target.userId === organization.ownerUserId) throw new InvalidOwnershipTransferError(); @@ -478,6 +545,7 @@ export async function transferOwnership(input: { userId: organization.ownerUserId, }, }, + include: { user: { select: { email: true } } }, }); if (!previous) throw new InvalidOwnershipTransferError(); await tx.membership.update({ @@ -488,9 +556,10 @@ export async function transferOwnership(input: { where: { id: target.id }, data: { role: "ORGANIZATION_OWNER", monthlySpendingCapCredits: null }, }); - await tx.organization.update({ + const ownership = await tx.organization.update({ where: { id: input.organizationId }, data: { ownerUserId: target.userId }, + select: { updatedAt: true }, }); await audit( tx, @@ -505,6 +574,31 @@ export async function transferOwnership(input: { targetMembershipId: target.id, }, ); + const eventVersion = ownership.updatedAt.getTime().toString(); + await enqueueMail( + teamMembershipChangedEmail({ + to: target.user.email, + organizationName: organization.name, + organizationId: input.organizationId, + userId: target.userId, + membershipId: target.id, + event: "OWNER_GRANTED", + eventVersion, + }), + tx, + ); + await enqueueMail( + teamMembershipChangedEmail({ + to: previous.user.email, + organizationName: organization.name, + organizationId: input.organizationId, + userId: previous.userId, + membershipId: previous.id, + event: "OWNER_RELEASED", + eventVersion, + }), + tx, + ); }); } @@ -649,7 +743,12 @@ export async function setUserPlatformRole(input: { await tx.$queryRaw`SELECT id FROM User WHERE id = ${input.targetUserId} FOR UPDATE`; const targetUser = await tx.user.findUnique({ where: { id: input.targetUserId }, - select: { id: true, platformRole: true, disabledAt: true }, + select: { + id: true, + email: true, + platformRole: true, + disabledAt: true, + }, }); if (!targetUser) throw new UserNotFoundError(); @@ -677,6 +776,19 @@ export async function setUserPlatformRole(input: { }, }, }); + await enqueueMail( + accountAdministrationEmail({ + to: targetUser.email, + userId: targetUser.id, + event: "PLATFORM_ROLE_CHANGED", + detail: `Your Aiwa Creators platform role changed from ${targetUser.platformRole} to ${updated.platformRole}.`, + eventVersion: entityVersion( + updated.updatedAt, + `role:${updated.platformRole}`, + ), + }), + tx, + ); return updated; }); @@ -698,7 +810,12 @@ export async function setUserDisabled(input: { await tx.$queryRaw`SELECT id FROM User WHERE id = ${input.targetUserId} FOR UPDATE`; const targetUser = await tx.user.findUnique({ where: { id: input.targetUserId }, - select: { id: true, platformRole: true, disabledAt: true }, + select: { + id: true, + email: true, + platformRole: true, + disabledAt: true, + }, }); if (!targetUser) throw new UserNotFoundError(); @@ -731,6 +848,21 @@ export async function setUserDisabled(input: { }, }, }); + await enqueueMail( + accountAdministrationEmail({ + to: targetUser.email, + userId: targetUser.id, + event: input.disabled ? "ACCOUNT_DISABLED" : "ACCOUNT_ENABLED", + detail: input.disabled + ? "An administrator disabled your Aiwa Creators account and active sessions were revoked." + : "An administrator reactivated your Aiwa Creators account.", + eventVersion: entityVersion( + updated.updatedAt, + `disabled:${input.disabled}`, + ), + }), + tx, + ); return updated; }); @@ -747,7 +879,7 @@ export async function revokeUserSessions(input: { return db.$transaction(async (tx) => { const targetUser = await tx.user.findUnique({ where: { id: input.targetUserId }, - select: { id: true }, + select: { id: true, email: true, updatedAt: true }, }); if (!targetUser) throw new UserNotFoundError(); @@ -769,6 +901,21 @@ export async function revokeUserSessions(input: { }, }, }); + if (deleted.count > 0) { + await enqueueMail( + accountAdministrationEmail({ + to: targetUser.email, + userId: targetUser.id, + event: "SESSIONS_REVOKED", + detail: + deleted.count === 1 + ? "An administrator revoked one active Aiwa Creators session." + : `An administrator revoked ${deleted.count} active Aiwa Creators sessions.`, + eventVersion: `${Date.now()}:${deleted.count}`, + }), + tx, + ); + } return deleted; }); @@ -844,7 +991,7 @@ export async function createOrganizationInvitation(input: { authorize(input.actor, input.organizationId, "members"); return db.$transaction( async (tx) => { - await lockedOrganization(tx, input.organizationId); + const organization = await lockedOrganization(tx, input.organizationId); const nonOwnerCount = await tx.membership.count({ where: { @@ -889,6 +1036,24 @@ export async function createOrganizationInvitation(input: { }, ); + if (normalizedEmail) { + const appUrl = process.env.APP_URL ?? "http://localhost:3000"; + await enqueueMail( + invitationEmail({ + to: normalizedEmail, + organizationName: organization.name, + organizationId: input.organizationId, + role: input.role, + invitationId: invitation.id, + invitationUrl: new URL( + `/invite/${encodeURIComponent(token)}`, + appUrl, + ).toString(), + }), + tx, + ); + } + return invitation; }, { isolationLevel: Prisma.TransactionIsolationLevel.Serializable }, diff --git a/packages/organizations/test/email-verification.test.ts b/packages/organizations/test/email-verification.test.ts index 66b1b7380..7ace0bd2a 100644 --- a/packages/organizations/test/email-verification.test.ts +++ b/packages/organizations/test/email-verification.test.ts @@ -1,4 +1,24 @@ -import { describe, expect, it, vi, beforeEach } from "vitest"; +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const { enqueueMailMock, teamMemberAddedEmailMock } = vi.hoisted(() => ({ + enqueueMailMock: vi.fn().mockResolvedValue({ id: "mail1", created: true }), + teamMemberAddedEmailMock: vi.fn((input) => ({ + kind: "SECURITY", + template: "organization.member_added.v1", + to: input.to, + subject: "member added", + text: "member added", + html: "

member added

", + organizationId: input.organizationId, + userId: input.userId, + idempotencyKey: `membership-added:${input.membershipId}`, + })), +})); + +vi.mock("@aiwa/mail", () => ({ + enqueueMail: enqueueMailMock, + teamMemberAddedEmail: teamMemberAddedEmailMock, +})); import { addMember, acceptOrganizationInvitation, @@ -59,6 +79,7 @@ describe("email verification enforcement", () => { id: "org1", status: "ACTIVE", ownerUserId: "owner1", + name: "Test Organization", }); mockTx.user.findUnique.mockResolvedValue({ id: "user1", @@ -88,9 +109,11 @@ describe("email verification enforcement", () => { id: "org1", status: "ACTIVE", ownerUserId: "owner1", + name: "Test Organization", }); mockTx.user.findUnique.mockResolvedValue({ id: "user2", + email: "verified@example.com", emailVerified: true, disabledAt: null, }); @@ -124,6 +147,19 @@ describe("email verification enforcement", () => { }), }), ); + expect(teamMemberAddedEmailMock).toHaveBeenCalledWith( + expect.objectContaining({ + to: "verified@example.com", + organizationName: "Test Organization", + membershipId: "mem1", + }), + ); + expect(enqueueMailMock).toHaveBeenCalledWith( + expect.objectContaining({ + idempotencyKey: "membership-added:mem1", + }), + mockTx, + ); }); }); diff --git a/packages/organizations/test/platform-owner-concurrency.test.ts b/packages/organizations/test/platform-owner-concurrency.test.ts index e3f5d4b68..814c151ea 100644 --- a/packages/organizations/test/platform-owner-concurrency.test.ts +++ b/packages/organizations/test/platform-owner-concurrency.test.ts @@ -20,6 +20,14 @@ const mockTx = { }, }; +vi.mock("@aiwa/mail", () => ({ + accountAdministrationEmail: vi.fn((input) => input), + enqueueMail: vi.fn().mockResolvedValue({ id: "mail-1", created: true }), + invitationEmail: vi.fn((input) => input), + teamMemberAddedEmail: vi.fn((input) => input), + teamMembershipChangedEmail: vi.fn((input) => input), +})); + vi.mock("@aiwa/db", () => ({ db: { $transaction: vi.fn(async (cb: (tx: typeof mockTx) => Promise) => diff --git a/packages/payments/package.json b/packages/payments/package.json index 1cf08789d..7279ea6b1 100644 --- a/packages/payments/package.json +++ b/packages/payments/package.json @@ -15,7 +15,8 @@ }, "dependencies": { "@aiwa/credits": "workspace:*", - "@aiwa/db": "workspace:*" + "@aiwa/db": "workspace:*", + "@aiwa/mail": "workspace:*" }, "devDependencies": { "vitest": "^4.1.11" diff --git a/packages/payments/src/index.ts b/packages/payments/src/index.ts index 889276084..56586b9e0 100644 --- a/packages/payments/src/index.ts +++ b/packages/payments/src/index.ts @@ -7,6 +7,7 @@ import { export type { LedgerEntry, ManualPayment } from "@aiwa/db"; import type { LedgerEntry, ManualPayment } from "@aiwa/db"; +import { billingStatusEmail, enqueueMail } from "@aiwa/mail"; const MAX_SIGNED_BIGINT = 9_223_372_036_854_775_807n; @@ -293,6 +294,26 @@ export async function _confirmPaymentTx( }, }); + const organization = await tx.organization.findUnique({ + where: { id: payment.organizationId }, + select: { name: true, owner: { select: { id: true, email: true } } }, + }); + if (organization) { + await enqueueMail( + billingStatusEmail({ + to: organization.owner.email, + organizationName: organization.name, + organizationId: payment.organizationId, + paymentId: payment.id, + status: "CONFIRMED", + amountBaisa: payment.amountBaisa, + userId: organization.owner.id, + detail: `${creditsGranted.toString()} credits were added to the workspace.`, + }), + tx, + ); + } + return { payment: updatedPayment, ledgerEntry }; } @@ -362,6 +383,26 @@ export async function _rejectPaymentTx( }, }); + const organization = await tx.organization.findUnique({ + where: { id: payment.organizationId }, + select: { name: true, owner: { select: { id: true, email: true } } }, + }); + if (organization) { + await enqueueMail( + billingStatusEmail({ + to: organization.owner.email, + organizationName: organization.name, + organizationId: payment.organizationId, + paymentId: payment.id, + status: "REJECTED", + amountBaisa: payment.amountBaisa, + userId: organization.owner.id, + detail: params.reason, + }), + tx, + ); + } + return updated; } @@ -480,6 +521,26 @@ export async function _reversePaymentTx( }, }); + const organization = await tx.organization.findUnique({ + where: { id: payment.organizationId }, + select: { name: true, owner: { select: { id: true, email: true } } }, + }); + if (organization) { + await enqueueMail( + billingStatusEmail({ + to: organization.owner.email, + organizationName: organization.name, + organizationId: payment.organizationId, + paymentId: payment.id, + status: "REVERSED", + amountBaisa: payment.amountBaisa, + userId: organization.owner.id, + detail: params.reason, + }), + tx, + ); + } + return { payment: updatedPayment, ledgerEntry }; } diff --git a/packages/payments/test/payments.test.ts b/packages/payments/test/payments.test.ts index e97f697f6..1d527b72d 100644 --- a/packages/payments/test/payments.test.ts +++ b/packages/payments/test/payments.test.ts @@ -1,4 +1,24 @@ import { describe, expect, it, vi } from "vitest"; + +const { enqueueMailMock, billingStatusEmailMock } = vi.hoisted(() => ({ + enqueueMailMock: vi.fn().mockResolvedValue({ id: "mail1", created: true }), + billingStatusEmailMock: vi.fn((input) => ({ + kind: "SECURITY", + template: "billing.payment.v1", + to: input.to, + subject: "billing status", + text: "billing status", + html: "

billing status

", + organizationId: input.organizationId, + userId: input.userId, + idempotencyKey: `billing:${input.paymentId}:${input.status.toLowerCase()}`, + })), +})); + +vi.mock("@aiwa/mail", () => ({ + enqueueMail: enqueueMailMock, + billingStatusEmail: billingStatusEmailMock, +})); import type { Prisma } from "@aiwa/db"; import { @@ -97,6 +117,16 @@ function createMockTx(initial?: { const tx = { $queryRaw: vi.fn().mockResolvedValue([]), + organization: { + findUnique: vi.fn(async ({ where }: { where: { id: string } }) => ({ + id: where.id, + name: "Test Organization", + owner: { + id: "owner_test_1", + email: "owner@example.com", + }, + })), + }, manualPayment: { findUnique: vi.fn( async ({ diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index b3445e52e..687dc5629 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -50,6 +50,9 @@ importers: '@aiwa/generation': specifier: workspace:* version: link:../../packages/generation + '@aiwa/mail': + specifier: workspace:* + version: link:../../packages/mail '@aiwa/organizations': specifier: workspace:* version: link:../../packages/organizations @@ -123,6 +126,9 @@ importers: '@aiwa/generation': specifier: workspace:* version: link:../../packages/generation + '@aiwa/mail': + specifier: workspace:* + version: link:../../packages/mail '@aiwa/providers': specifier: workspace:* version: link:../../packages/providers @@ -211,6 +217,9 @@ importers: '@aiwa/db': specifier: workspace:* version: link:../db + '@aiwa/mail': + specifier: workspace:* + version: link:../mail '@aiwa/organizations': specifier: workspace:* version: link:../organizations @@ -225,6 +234,25 @@ importers: specifier: ^4.1.11 version: 4.1.11(@types/node@24.13.4)(vite@7.3.6(@types/node@24.13.4)(jiti@2.7.0)(lightningcss@1.32.0)(tsx@4.23.13)) + packages/mail: + dependencies: + '@aiwa/config': + specifier: workspace:* + version: link:../config + '@aiwa/db': + specifier: workspace:* + version: link:../db + nodemailer: + specifier: 9.1.1 + version: 9.1.1 + devDependencies: + '@types/nodemailer': + specifier: 8.0.2 + version: 8.0.2 + vitest: + specifier: ^4.1.11 + version: 4.1.11(@types/node@24.13.4)(vite@7.3.6(@types/node@24.13.4)(jiti@2.7.0)(lightningcss@1.32.0)(tsx@4.23.13)) + packages/observability: devDependencies: vitest: @@ -239,6 +267,9 @@ importers: '@aiwa/db': specifier: workspace:* version: link:../db + '@aiwa/mail': + specifier: workspace:* + version: link:../mail devDependencies: vitest: specifier: ^4.1.11 @@ -252,6 +283,9 @@ importers: '@aiwa/db': specifier: workspace:* version: link:../db + '@aiwa/mail': + specifier: workspace:* + version: link:../mail devDependencies: vitest: specifier: ^4.1.11 @@ -1512,6 +1546,9 @@ packages: '@types/node@24.13.4': resolution: {integrity: sha512-YJ7EqCstVTzIr0fMr7qul/977en+pQHrfmuKIo6Zr9i75Be21dr3MovcfvGtyvi2HAUrRerWps5sMO9I7WaxDw==} + '@types/nodemailer@8.0.2': + resolution: {integrity: sha512-c7M5ox8p0nEOfbJ2E9Qcmt8/QCu/VzsiAhzBiZbvky2PhKZDHpKB3sQHBM5MEZNkCfBOaS6S9//AGPRTNB7IaA==} + '@types/react-dom@19.3.0': resolution: {integrity: sha512-ZI7bU42mZXXKHn/qNLEw2IrbiINU7X5+vfgdixBHkCNpYWXjKgfQ/P+uyGb5CjOLB9UcnTeg3rylQtV2hym44Q==} peerDependencies: @@ -2875,6 +2912,10 @@ packages: resolution: {integrity: sha512-mIrE/Cw9y+9Au6dS5vDKDhQza9YvG6w+ZrS6X+ZzA7yFW/soAeaups4Qzn1bL6g5FVy8WtP79+0j82oPIbqRjQ==} engines: {node: '>=18'} + nodemailer@9.1.1: + resolution: {integrity: sha512-izw9mVKFix6YSnC9eLgV6g1opl9DUlRio9ZNcq+Wu9Ujn2UwF+8Nl0B8nz22kEC+CTZCvinkxwJ0DeFbb6NwcQ==} + engines: {node: '>=6.0.0'} + nypm@0.6.10: resolution: {integrity: sha512-W72Hrj1petq+b3Hk2aAC+9zswetlIFTnDW4s5djseZh2nYqBbyQLOtj472HwcbcWykPBUW1WpWpjOd4nK6gMRw==} engines: {node: '>=18'} @@ -4389,6 +4430,10 @@ snapshots: dependencies: undici-types: 7.18.2 + '@types/nodemailer@8.0.2': + dependencies: + '@types/node': 24.13.4 + '@types/react-dom@19.3.0(@types/react@19.3.0)': dependencies: '@types/react': 19.3.0 @@ -5902,6 +5947,8 @@ snapshots: node-releases@2.0.55: {} + nodemailer@9.1.1: {} + nypm@0.6.10: dependencies: citty: 0.2.2