✨ Added new procedure for sending admin email (#1312)

* 🚧 Added new lexicon for sending admin email

* ✨ Add moderation mailer

* ✨ Switch to text email content from html

* 🧹 Cleanup some early implementation code and reflect PR reivew

* ✨ Use smtp host instead of gmail service config

* ✨ Move to using single smtp url

lexicons/com/atproto/admin/sendEmail.json

@@ -0,0 +1,32 @@
+{
+  "lexicon": 1,
+  "id": "com.atproto.admin.sendEmail",
+  "defs": {
+    "main": {
+      "type": "procedure",
+      "description": "Send email to a user's primary email address",
+      "input": {
+        "encoding": "application/json",
+        "schema": {
+          "type": "object",
+          "required": ["recipientDid", "content"],
+          "properties": {
+            "recipientDid": { "type": "string", "format": "did" },
+            "content": { "type": "string" },
+            "subject": { "type": "string" }
+          }
+        }
+      },
+      "output": {
+        "encoding": "application/json",
+        "schema": {
+          "type": "object",
+          "required": ["sent"],
+          "properties": {
+            "sent": { "type": "boolean" }
+          }
+        }
+      }
+    }
+  }
+}

packages/api/src/client/index.ts

@@ -22,6 +22,7 @@ import * as ComAtprotoAdminRebaseRepo from './types/com/atproto/admin/rebaseRepo
 import * as ComAtprotoAdminResolveModerationReports from './types/com/atproto/admin/resolveModerationReports'
 import * as ComAtprotoAdminReverseModerationAction from './types/com/atproto/admin/reverseModerationAction'
 import * as ComAtprotoAdminSearchRepos from './types/com/atproto/admin/searchRepos'
+import * as ComAtprotoAdminSendEmail from './types/com/atproto/admin/sendEmail'
 import * as ComAtprotoAdminTakeModerationAction from './types/com/atproto/admin/takeModerationAction'
 import * as ComAtprotoAdminUpdateAccountEmail from './types/com/atproto/admin/updateAccountEmail'
 import * as ComAtprotoAdminUpdateAccountHandle from './types/com/atproto/admin/updateAccountHandle'
@@ -140,6 +141,7 @@ export * as ComAtprotoAdminRebaseRepo from './types/com/atproto/admin/rebaseRepo
 export * as ComAtprotoAdminResolveModerationReports from './types/com/atproto/admin/resolveModerationReports'
 export * as ComAtprotoAdminReverseModerationAction from './types/com/atproto/admin/reverseModerationAction'
 export * as ComAtprotoAdminSearchRepos from './types/com/atproto/admin/searchRepos'
+export * as ComAtprotoAdminSendEmail from './types/com/atproto/admin/sendEmail'
 export * as ComAtprotoAdminTakeModerationAction from './types/com/atproto/admin/takeModerationAction'
 export * as ComAtprotoAdminUpdateAccountEmail from './types/com/atproto/admin/updateAccountEmail'
 export * as ComAtprotoAdminUpdateAccountHandle from './types/com/atproto/admin/updateAccountHandle'
@@ -484,6 +486,17 @@ export class AdminNS {
       })
   }
 
+  sendEmail(
+    data?: ComAtprotoAdminSendEmail.InputSchema,
+    opts?: ComAtprotoAdminSendEmail.CallOptions,
+  ): Promise<ComAtprotoAdminSendEmail.Response> {
+    return this._service.xrpc
+      .call('com.atproto.admin.sendEmail', opts?.qp, data, opts)
+      .catch((e) => {
+        throw ComAtprotoAdminSendEmail.toKnownErr(e)
+      })
+  }
+
   takeModerationAction(
     data?: ComAtprotoAdminTakeModerationAction.InputSchema,
     opts?: ComAtprotoAdminTakeModerationAction.CallOptions,

packages/api/src/client/lexicons.ts

@@ -1158,6 +1158,47 @@ export const schemaDict = {
       },
     },
   },
+  ComAtprotoAdminSendEmail: {
+    lexicon: 1,
+    id: 'com.atproto.admin.sendEmail',
+    defs: {
+      main: {
+        type: 'procedure',
+        description: "Send email to a user's primary email address",
+        input: {
+          encoding: 'application/json',
+          schema: {
+            type: 'object',
+            required: ['recipientDid', 'content'],
+            properties: {
+              recipientDid: {
+                type: 'string',
+                format: 'did',
+              },
+              content: {
+                type: 'string',
+              },
+              subject: {
+                type: 'string',
+              },
+            },
+          },
+        },
+        output: {
+          encoding: 'application/json',
+          schema: {
+            type: 'object',
+            required: ['sent'],
+            properties: {
+              sent: {
+                type: 'boolean',
+              },
+            },
+          },
+        },
+      },
+    },
+  },
   ComAtprotoAdminTakeModerationAction: {
     lexicon: 1,
     id: 'com.atproto.admin.takeModerationAction',
@@ -6397,6 +6438,7 @@ export const ids = {
   ComAtprotoAdminReverseModerationAction:
     'com.atproto.admin.reverseModerationAction',
   ComAtprotoAdminSearchRepos: 'com.atproto.admin.searchRepos',
+  ComAtprotoAdminSendEmail: 'com.atproto.admin.sendEmail',
   ComAtprotoAdminTakeModerationAction: 'com.atproto.admin.takeModerationAction',
   ComAtprotoAdminUpdateAccountEmail: 'com.atproto.admin.updateAccountEmail',
   ComAtprotoAdminUpdateAccountHandle: 'com.atproto.admin.updateAccountHandle',

packages/api/src/client/types/com/atproto/admin/sendEmail.ts

@@ -0,0 +1,40 @@
+/**
+ * GENERATED CODE - DO NOT MODIFY
+ */
+import { Headers, XRPCError } from '@atproto/xrpc'
+import { ValidationResult, BlobRef } from '@atproto/lexicon'
+import { isObj, hasProp } from '../../../../util'
+import { lexicons } from '../../../../lexicons'
+import { CID } from 'multiformats/cid'
+
+export interface QueryParams {}
+
+export interface InputSchema {
+  recipientDid: string
+  content: string
+  subject?: string
+  [k: string]: unknown
+}
+
+export interface OutputSchema {
+  sent: boolean
+  [k: string]: unknown
+}
+
+export interface CallOptions {
+  headers?: Headers
+  qp?: QueryParams
+  encoding: 'application/json'
+}
+
+export interface Response {
+  success: boolean
+  headers: Headers
+  data: OutputSchema
+}
+
+export function toKnownErr(e: any) {
+  if (e instanceof XRPCError) {
+  }
+  return e
+}

packages/bsky/src/lexicon/index.ts

@@ -23,6 +23,7 @@ import * as ComAtprotoAdminRebaseRepo from './types/com/atproto/admin/rebaseRepo
 import * as ComAtprotoAdminResolveModerationReports from './types/com/atproto/admin/resolveModerationReports'
 import * as ComAtprotoAdminReverseModerationAction from './types/com/atproto/admin/reverseModerationAction'
 import * as ComAtprotoAdminSearchRepos from './types/com/atproto/admin/searchRepos'
+import * as ComAtprotoAdminSendEmail from './types/com/atproto/admin/sendEmail'
 import * as ComAtprotoAdminTakeModerationAction from './types/com/atproto/admin/takeModerationAction'
 import * as ComAtprotoAdminUpdateAccountEmail from './types/com/atproto/admin/updateAccountEmail'
 import * as ComAtprotoAdminUpdateAccountHandle from './types/com/atproto/admin/updateAccountHandle'
@@ -303,6 +304,13 @@ export class AdminNS {
     return this._server.xrpc.method(nsid, cfg)
   }
 
+  sendEmail<AV extends AuthVerifier>(
+    cfg: ConfigOf<AV, ComAtprotoAdminSendEmail.Handler<ExtractAuth<AV>>>,
+  ) {
+    const nsid = 'com.atproto.admin.sendEmail' // @ts-ignore
+    return this._server.xrpc.method(nsid, cfg)
+  }
+
   takeModerationAction<AV extends AuthVerifier>(
     cfg: ConfigOf<
       AV,

packages/bsky/src/lexicon/lexicons.ts

@@ -1158,6 +1158,47 @@ export const schemaDict = {
       },
     },
   },
+  ComAtprotoAdminSendEmail: {
+    lexicon: 1,
+    id: 'com.atproto.admin.sendEmail',
+    defs: {
+      main: {
+        type: 'procedure',
+        description: "Send email to a user's primary email address",
+        input: {
+          encoding: 'application/json',
+          schema: {
+            type: 'object',
+            required: ['recipientDid', 'content'],
+            properties: {
+              recipientDid: {
+                type: 'string',
+                format: 'did',
+              },
+              content: {
+                type: 'string',
+              },
+              subject: {
+                type: 'string',
+              },
+            },
+          },
+        },
+        output: {
+          encoding: 'application/json',
+          schema: {
+            type: 'object',
+            required: ['sent'],
+            properties: {
+              sent: {
+                type: 'boolean',
+              },
+            },
+          },
+        },
+      },
+    },
+  },
   ComAtprotoAdminTakeModerationAction: {
     lexicon: 1,
     id: 'com.atproto.admin.takeModerationAction',
@@ -6397,6 +6438,7 @@ export const ids = {
   ComAtprotoAdminReverseModerationAction:
     'com.atproto.admin.reverseModerationAction',
   ComAtprotoAdminSearchRepos: 'com.atproto.admin.searchRepos',
+  ComAtprotoAdminSendEmail: 'com.atproto.admin.sendEmail',
   ComAtprotoAdminTakeModerationAction: 'com.atproto.admin.takeModerationAction',
   ComAtprotoAdminUpdateAccountEmail: 'com.atproto.admin.updateAccountEmail',
   ComAtprotoAdminUpdateAccountHandle: 'com.atproto.admin.updateAccountHandle',

packages/bsky/src/lexicon/types/com/atproto/admin/sendEmail.ts

@@ -0,0 +1,48 @@
+/**
+ * GENERATED CODE - DO NOT MODIFY
+ */
+import express from 'express'
+import { ValidationResult, BlobRef } from '@atproto/lexicon'
+import { lexicons } from '../../../../lexicons'
+import { isObj, hasProp } from '../../../../util'
+import { CID } from 'multiformats/cid'
+import { HandlerAuth } from '@atproto/xrpc-server'
+
+export interface QueryParams {}
+
+export interface InputSchema {
+  recipientDid: string
+  content: string
+  subject?: string
+  [k: string]: unknown
+}
+
+export interface OutputSchema {
+  sent: boolean
+  [k: string]: unknown
+}
+
+export interface HandlerInput {
+  encoding: 'application/json'
+  body: InputSchema
+}
+
+export interface HandlerSuccess {
+  encoding: 'application/json'
+  body: OutputSchema
+  headers?: { [key: string]: string }
+}
+
+export interface HandlerError {
+  status: number
+  message?: string
+}
+
+export type HandlerOutput = HandlerError | HandlerSuccess
+export type Handler<HA extends HandlerAuth = never> = (ctx: {
+  auth: HA
+  params: QueryParams
+  input: HandlerInput
+  req: express.Request
+  res: express.Response
+}) => Promise<HandlerOutput> | HandlerOutput

packages/pds/src/api/com/atproto/admin/index.ts

@@ -17,6 +17,7 @@ import getInviteCodes from './getInviteCodes'
 import updateAccountHandle from './updateAccountHandle'
 import updateAccountEmail from './updateAccountEmail'
 import rebaseRepo from './rebaseRepo'
+import sendEmail from './sendEmail'
 
 export default function (server: Server, ctx: AppContext) {
   resolveModerationReports(server, ctx)
@@ -36,4 +37,5 @@ export default function (server: Server, ctx: AppContext) {
   updateAccountHandle(server, ctx)
   updateAccountEmail(server, ctx)
   rebaseRepo(server, ctx)
+  sendEmail(server, ctx)
 }

packages/pds/src/api/com/atproto/admin/sendEmail.ts

@@ -0,0 +1,38 @@
+import { AuthRequiredError, InvalidRequestError } from '@atproto/xrpc-server'
+import { Server } from '../../../../lexicon'
+import AppContext from '../../../../context'
+
+export default function (server: Server, ctx: AppContext) {
+  server.com.atproto.admin.sendEmail({
+    auth: ctx.roleVerifier,
+    handler: async ({ input, auth }) => {
+      if (!auth.credentials.admin && !auth.credentials.moderator) {
+        throw new AuthRequiredError('Insufficient privileges')
+      }
+
+      const {
+        content,
+        recipientDid,
+        subject = 'Message from Bluesky moderator',
+      } = input.body
+      const userInfo = await ctx.db.db
+        .selectFrom('user_account')
+        .where('did', '=', recipientDid)
+        .select('email')
+        .executeTakeFirst()
+
+      if (!userInfo) {
+        throw new InvalidRequestError('Recipient not found')
+      }
+
+      await ctx.moderationMailer.send(
+        { content },
+        { subject, to: userInfo.email },
+      )
+      return {
+        encoding: 'application/json',
+        body: { sent: true },
+      }
+    },
+  })
+}

packages/pds/src/config.ts

@@ -46,6 +46,8 @@ export interface ServerConfigValues {
   appUrlPasswordReset: string
   emailSmtpUrl?: string
   emailNoReplyAddress: string
+  moderationEmailAddress?: string
+  moderationEmailSmtpUrl?: string
 
   hiveApiKey?: string
   labelerDid: string
@@ -162,6 +164,11 @@ export class ServerConfig {
     const emailNoReplyAddress =
       process.env.EMAIL_NO_REPLY_ADDRESS || 'noreply@blueskyweb.xyz'
 
+    const moderationEmailAddress =
+      process.env.MODERATION_EMAIL_ADDRESS || undefined
+    const moderationEmailSmtpUrl =
+      process.env.MODERATION_EMAIL_SMTP_URL || undefined
+
     const hiveApiKey = process.env.HIVE_API_KEY || undefined
     const labelerDid = process.env.LABELER_DID || 'did:example:labeler'
     const labelerKeywords = {}
@@ -247,6 +254,8 @@ export class ServerConfig {
       appUrlPasswordReset,
       emailSmtpUrl,
       emailNoReplyAddress,
+      moderationEmailAddress,
+      moderationEmailSmtpUrl,
       hiveApiKey,
       labelerDid,
       labelerKeywords,
@@ -432,6 +441,14 @@ export class ServerConfig {
     return this.cfg.emailNoReplyAddress
   }
 
+  get moderationEmailAddress() {
+    return this.cfg.moderationEmailAddress
+  }
+
+  get moderationEmailSmtpUrl() {
+    return this.cfg.moderationEmailSmtpUrl
+  }
+
   get hiveApiKey() {
     return this.cfg.hiveApiKey
   }

packages/pds/src/context.ts

@@ -8,6 +8,7 @@ import { Database } from './db'
 import { ServerConfig } from './config'
 import * as auth from './auth'
 import { ServerMailer } from './mailer'
+import { ModerationMailer } from './mailer/moderation'
 import { BlobStore } from '@atproto/repo'
 import { ImageUriBuilder } from './image/uri'
 import { Services } from './services'
@@ -36,6 +37,7 @@ export class AppContext {
       imgUriBuilder: ImageUriBuilder
       cfg: ServerConfig
       mailer: ServerMailer
+      moderationMailer: ModerationMailer
       services: Services
       messageDispatcher: MessageDispatcher
       sequencer: Sequencer
@@ -111,6 +113,10 @@ export class AppContext {
     return this.opts.mailer
   }
 
+  get moderationMailer(): ModerationMailer {
+    return this.opts.moderationMailer
+  }
+
   get services(): Services {
     return this.opts.services
   }

packages/pds/src/index.ts

@@ -24,6 +24,7 @@ import compression from './util/compression'
 import { dbLogger, loggerMiddleware, seqLogger } from './logger'
 import { ServerConfig } from './config'
 import { ServerMailer } from './mailer'
+import { ModerationMailer } from './mailer/moderation'
 import { createServer } from './lexicon'
 import { MessageDispatcher } from './event-stream/message-queue'
 import { ImageUriBuilder } from './image/uri'
@@ -108,12 +109,21 @@ export class PDS {
       ? new SequencerLeader(db, config.sequencerLeaderLockId)
       : null
 
-    const mailTransport =
+    const serverMailTransport =
       config.emailSmtpUrl !== undefined
         ? createTransport(config.emailSmtpUrl)
         : createTransport({ jsonTransport: true })
 
-    const mailer = new ServerMailer(mailTransport, config)
+    const moderationMailTransport =
+      config.moderationEmailSmtpUrl !== undefined
+        ? createTransport(config.moderationEmailSmtpUrl)
+        : createTransport({ jsonTransport: true })
+
+    const mailer = new ServerMailer(serverMailTransport, config)
+    const moderationMailer = new ModerationMailer(
+      moderationMailTransport,
+      config,
+    )
 
     const app = express()
     app.use(cors())
@@ -224,6 +234,7 @@ export class PDS {
       contentReporter,
       services,
       mailer,
+      moderationMailer,
       imgUriBuilder,
       backgroundQueue,
       crawlers,

packages/pds/src/lexicon/index.ts

@@ -23,6 +23,7 @@ import * as ComAtprotoAdminRebaseRepo from './types/com/atproto/admin/rebaseRepo
 import * as ComAtprotoAdminResolveModerationReports from './types/com/atproto/admin/resolveModerationReports'
 import * as ComAtprotoAdminReverseModerationAction from './types/com/atproto/admin/reverseModerationAction'
 import * as ComAtprotoAdminSearchRepos from './types/com/atproto/admin/searchRepos'
+import * as ComAtprotoAdminSendEmail from './types/com/atproto/admin/sendEmail'
 import * as ComAtprotoAdminTakeModerationAction from './types/com/atproto/admin/takeModerationAction'
 import * as ComAtprotoAdminUpdateAccountEmail from './types/com/atproto/admin/updateAccountEmail'
 import * as ComAtprotoAdminUpdateAccountHandle from './types/com/atproto/admin/updateAccountHandle'
@@ -303,6 +304,13 @@ export class AdminNS {
     return this._server.xrpc.method(nsid, cfg)
   }
 
+  sendEmail<AV extends AuthVerifier>(
+    cfg: ConfigOf<AV, ComAtprotoAdminSendEmail.Handler<ExtractAuth<AV>>>,
+  ) {
+    const nsid = 'com.atproto.admin.sendEmail' // @ts-ignore
+    return this._server.xrpc.method(nsid, cfg)
+  }
+
   takeModerationAction<AV extends AuthVerifier>(
     cfg: ConfigOf<
       AV,

packages/pds/src/lexicon/lexicons.ts

@@ -1158,6 +1158,47 @@ export const schemaDict = {
       },
     },
   },
+  ComAtprotoAdminSendEmail: {
+    lexicon: 1,
+    id: 'com.atproto.admin.sendEmail',
+    defs: {
+      main: {
+        type: 'procedure',
+        description: "Send email to a user's primary email address",
+        input: {
+          encoding: 'application/json',
+          schema: {
+            type: 'object',
+            required: ['recipientDid', 'content'],
+            properties: {
+              recipientDid: {
+                type: 'string',
+                format: 'did',
+              },
+              content: {
+                type: 'string',
+              },
+              subject: {
+                type: 'string',
+              },
+            },
+          },
+        },
+        output: {
+          encoding: 'application/json',
+          schema: {
+            type: 'object',
+            required: ['sent'],
+            properties: {
+              sent: {
+                type: 'boolean',
+              },
+            },
+          },
+        },
+      },
+    },
+  },
   ComAtprotoAdminTakeModerationAction: {
     lexicon: 1,
     id: 'com.atproto.admin.takeModerationAction',
@@ -6397,6 +6438,7 @@ export const ids = {
   ComAtprotoAdminReverseModerationAction:
     'com.atproto.admin.reverseModerationAction',
   ComAtprotoAdminSearchRepos: 'com.atproto.admin.searchRepos',
+  ComAtprotoAdminSendEmail: 'com.atproto.admin.sendEmail',
   ComAtprotoAdminTakeModerationAction: 'com.atproto.admin.takeModerationAction',
   ComAtprotoAdminUpdateAccountEmail: 'com.atproto.admin.updateAccountEmail',
   ComAtprotoAdminUpdateAccountHandle: 'com.atproto.admin.updateAccountHandle',

packages/pds/src/lexicon/types/com/atproto/admin/sendEmail.ts

@@ -0,0 +1,48 @@
+/**
+ * GENERATED CODE - DO NOT MODIFY
+ */
+import express from 'express'
+import { ValidationResult, BlobRef } from '@atproto/lexicon'
+import { lexicons } from '../../../../lexicons'
+import { isObj, hasProp } from '../../../../util'
+import { CID } from 'multiformats/cid'
+import { HandlerAuth } from '@atproto/xrpc-server'
+
+export interface QueryParams {}
+
+export interface InputSchema {
+  recipientDid: string
+  content: string
+  subject?: string
+  [k: string]: unknown
+}
+
+export interface OutputSchema {
+  sent: boolean
+  [k: string]: unknown
+}
+
+export interface HandlerInput {
+  encoding: 'application/json'
+  body: InputSchema
+}
+
+export interface HandlerSuccess {
+  encoding: 'application/json'
+  body: OutputSchema
+  headers?: { [key: string]: string }
+}
+
+export interface HandlerError {
+  status: number
+  message?: string
+}
+
+export type HandlerOutput = HandlerError | HandlerSuccess
+export type Handler<HA extends HandlerAuth = never> = (ctx: {
+  auth: HA
+  params: QueryParams
+  input: HandlerInput
+  req: express.Request
+  res: express.Response
+}) => Promise<HandlerOutput> | HandlerOutput

packages/pds/src/mailer/moderation.ts

@@ -0,0 +1,34 @@
+import { Transporter } from 'nodemailer'
+import Mail from 'nodemailer/lib/mailer'
+import SMTPTransport from 'nodemailer/lib/smtp-transport'
+import { ServerConfig } from '../config'
+import { mailerLogger } from '../logger'
+
+export class ModerationMailer {
+  private config: ServerConfig
+  transporter: Transporter<SMTPTransport.SentMessageInfo>
+
+  constructor(
+    transporter: Transporter<SMTPTransport.SentMessageInfo>,
+    config: ServerConfig,
+  ) {
+    this.config = config
+    this.transporter = transporter
+  }
+
+  async send({ content }: { content: string }, mailOpts: Mail.Options) {
+    const res = await this.transporter.sendMail({
+      ...mailOpts,
+      text: content,
+      from: this.config.moderationEmailAddress,
+    })
+
+    if (!this.config.moderationEmailSmtpUrl) {
+      mailerLogger.debug(
+        'Moderation email auth is not configured. Intended to send email:\n' +
+          JSON.stringify(res, null, 2),
+      )
+    }
+    return res
+  }
+}