feat: built-in job for Mail.queue of dox_mailer

CHANGELOG.md

@@ -1,5 +1,8 @@
 ## Unreleased
 
+- New: a built-in job for `Mail.queue()` and `Mail.later()` of dox_mailer. `Queue.configure()` turns it on. Apps do not add it to `lib/generated/jobs.g.dart`: every `JobRegistry` finds the built-in jobs (`JobRegistry.builtIn`) after its own jobs. With `encrypt: true`, the mail is saved like a `ShouldBeEncrypted` job.
+- Now depends on dox_mailer.
+- Deploy the queue worker together with the web app. An old worker does not know the new mail job, so it fails the queued emails.
 - Fix (opt-in): a job that timed out keeps running, and its retry could run at the same time. The new worker flag `--timeout-grace=SECONDS` (`WorkerOptions.timeoutGrace`) makes the worker wait for the old run before the next try. The default does not change. For the in-process worker, use `QueueConfig(workerTimeoutGrace: ...)`.
 - New opt-in `QueueConfig.redisFromEnv()`: reads the Redis server from `REDIS_URL`, or from `REDIS_HOST`, `REDIS_PORT`, `REDIS_USERNAME`, `REDIS_PASSWORD` and `REDIS_TLS`. Use it as `QueueConfig(redis: QueueConfig.redisFromEnv())`. Without it, the default is still localhost:6379.
 - Opt-in `ShouldBeEncrypted` job mixin: the job data is saved encrypted with the app key (AES-GCM of dox_core). Jobs without it are saved as before.

README.md

@@ -264,6 +264,19 @@ Run a job now, in this process, with no queue:
 await SendWelcomeEmail(userId: 1).dispatchSync();
 ```
 
+### Queue an email
+
+You do not need a job class to send an email later. With [dox_mailer](https://pub.dev/packages/dox_mailer#queue-an-email), call `Mail.queue()`:
+
+```dart
+await Mail.queue(message);
+await Mail.later(Duration(minutes: 10), message, queue: 'emails');
+```
+
+dox_queue has a built-in job for it. It is not in `lib/generated/jobs.g.dart`, and you do not need to add it.
+
+Deploy the worker together with the web app. An old worker does not know this job, so it fails the queued emails.
+
 ## Run the worker
 
 ```bash

ci/use-siblings.sh

@@ -33,6 +33,7 @@ siblings=(
   "dox-orm dox_orm:packages/dox-orm dox_migration:packages/dox-migration"
   "dox-postgres-native dox_postgres_native:."
   "dox-mysql-native dox_mysql_native:."
+  "dox-mailer dox_mailer:."
 )
 
 if [ "${GITHUB_REF_TYPE:-}" = tag ]; then

lib/src/builder/job_registry_builder.dart

@@ -100,6 +100,12 @@ class JobRegistryBuilder implements Builder {
       );
     }
     String name = job.getField('name')?.toStringValue() ?? className;
+    if (name.startsWith('dox_queue:')) {
+      throw StateError(
+        '@Job() on $where: the name "$name" starts with "dox_queue:". '
+        'That start is kept for the built-in jobs. Use another name.',
+      );
+    }
     JobEntry entry = JobEntry(element, library, name);
     for (FormalParameterElement param in constructor.formalParameters) {
       String field = param.displayName;

lib/src/job_registry.dart

@@ -1,4 +1,5 @@
 import 'job.dart';
+import 'send_queued_mail.dart';
 
 /// Turns a job into a JSON map (its constructor parameters).
 typedef JobEncoder = Map<String, Object?> Function(DoxJob job);
@@ -24,18 +25,26 @@ class JobRegistry {
 
   final List<JobCodec> codecs;
 
+  /// The jobs of dox_queue itself, eg. the job of `Mail.queue()`. Every
+  /// registry finds them after its own [codecs], so an app does not add
+  /// them.
+  static const List<JobCodec> builtIn = <JobCodec>[
+    SendQueuedMail.codec,
+    SendEncryptedQueuedMail.codec,
+  ];
+
   /// the codec of a job class, or null
   JobCodec? byType(Type type) {
     // ponytail: a list scan, an app has few job classes
-    for (JobCodec codec in codecs) {
+    for (JobCodec codec in codecs.followedBy(builtIn)) {
       if (codec.type == type) return codec;
     }
     return null;
   }
 
   /// the codec of a saved job name, or null
   JobCodec? byName(String name) {
-    for (JobCodec codec in codecs) {
+    for (JobCodec codec in codecs.followedBy(builtIn)) {
       if (codec.name == name) return codec;
     }
     return null;

lib/src/queue.dart

@@ -1,11 +1,13 @@
 import 'dart:convert';
 
 import 'package:dox_core/dox_core.dart' show AESEncryptor, Dox;
+import 'package:dox_mailer/dox_mailer.dart' show Mail;
 
 import 'job.dart';
 import 'job_registry.dart';
 import 'queue_config.dart';
 import 'queue_driver.dart';
+import 'send_queued_mail.dart';
 
 /// Puts jobs on the queue. `QueueProvider` sets it up in every isolate.
 /// ```
@@ -22,9 +24,12 @@ class Queue {
 
   /// Use [config] in this isolate. `QueueProvider` calls it. Call it in
   /// tests, eg. with the `sync` driver.
+  ///
+  /// It also lets `Mail.queue()` of dox_mailer put mails on this queue.
   static void configure(QueueConfig config) {
     _config = config;
     _driver = null;
+    Mail.useQueue(const QueueMailQueue());
   }
 
   /// the config from [configure]

lib/src/send_queued_mail.dart

@@ -0,0 +1,94 @@
+import 'package:dox_mailer/dox_mailer.dart' show Mail, MailQueue, Message;
+
+import 'job.dart';
+import 'job_registry.dart';
+import 'queue.dart';
+
+/// The built-in job of `Mail.queue()` and `Mail.later()` of dox_mailer.
+/// It sends a saved [message] with the mailer [mailer].
+///
+/// Apps do not use it. It is not in `lib/generated/jobs.g.dart`: the
+/// [JobRegistry] finds it by itself (see [JobRegistry.builtIn]).
+class SendQueuedMail extends DoxJob {
+  SendQueuedMail({required this.mailer, required this.message});
+
+  /// the mailer name of the mail config, eg. `smtp`
+  final String mailer;
+
+  /// the message, from `Message.toJson()`
+  final Map<String, Object?> message;
+
+  /// the saved name. The `dox_queue:` start can not clash with an app job.
+  static const String jobName = 'dox_queue:SendQueuedMail';
+
+  static const JobCodec codec = JobCodec(
+    jobName,
+    SendQueuedMail,
+    encode,
+    _decode,
+  );
+
+  @override
+  Future<void> handle() async {
+    await Mail.mailer(mailer).send(Message.fromJson(message));
+  }
+
+  static Map<String, Object?> encode(DoxJob job) {
+    SendQueuedMail value = job as SendQueuedMail;
+    return <String, Object?>{'mailer': value.mailer, 'message': value.message};
+  }
+
+  static DoxJob _decode(Map<String, Object?> data) {
+    return SendQueuedMail(
+      mailer: data['mailer'] as String,
+      message: Map<String, Object?>.from(
+        data['message'] as Map<Object?, Object?>,
+      ),
+    );
+  }
+}
+
+/// [SendQueuedMail] with its data saved encrypted, for
+/// `Mail.queue(message, encrypt: true)`.
+class SendEncryptedQueuedMail extends SendQueuedMail with ShouldBeEncrypted {
+  SendEncryptedQueuedMail({required super.mailer, required super.message});
+
+  static const String jobName = 'dox_queue:SendEncryptedQueuedMail';
+
+  static const JobCodec codec = JobCodec(
+    jobName,
+    SendEncryptedQueuedMail,
+    SendQueuedMail.encode,
+    _decode,
+  );
+
+  static DoxJob _decode(Map<String, Object?> data) {
+    return SendEncryptedQueuedMail(
+      mailer: data['mailer'] as String,
+      message: Map<String, Object?>.from(
+        data['message'] as Map<Object?, Object?>,
+      ),
+    );
+  }
+}
+
+/// Puts the mails of `Mail.queue()` on this queue. `Queue.configure()`
+/// sets it with `Mail.useQueue()`.
+class QueueMailQueue implements MailQueue {
+  const QueueMailQueue();
+
+  @override
+  Future<String> push(
+    String mailer,
+    Message message, {
+    Duration? delay,
+    String? queue,
+    bool encrypt = false,
+  }) {
+    Map<String, Object?> data = message.toJson();
+    DoxJob job = encrypt
+        ? SendEncryptedQueuedMail(mailer: mailer, message: data)
+        : SendQueuedMail(mailer: mailer, message: data);
+    return Queue.dispatch(job, delay: delay, queue: queue);
+  }
+}

pubspec.yaml

@@ -12,6 +12,7 @@ dependencies:
   build: ^5.0.0
   dart_style: ^3.1.14
   dox_core: ^3.0.2
+  dox_mailer: ^3.0.0
   dox_orm: ^3.0.0
   glob: ^2.2.0
   ioredis: ^3.0.1

test/builder_test.dart

@@ -194,4 +194,16 @@ class B extends DoxJob {
 ''');
     expect(error, contains('Two jobs are named "same"'));
   });
+
+  test('a job name can not start with dox_queue:', () async {
+    String error = await _error('''
+$_header
+@Job(name: 'dox_queue:SendQueuedMail')
+class A extends DoxJob {
+  @override
+  Future<void> handle() async {}
+}
+''');
+    expect(error, contains('kept for the built-in jobs'));
+  });
 }

test/queue_mail_test.dart

@@ -0,0 +1,137 @@
+import 'package:dox_mailer/dox_mailer.dart';
+import 'package:dox_queue/dox_queue.dart';
+import 'package:test/test.dart';
+
+import 'support.dart';
+
+late ArrayTransport mainMailer;
+late ArrayTransport otherMailer;
+
+/// a mail server that is down
+class BrokenTransport extends Transport {
+  int calls = 0;
+
+  @override
+  Future<SentMessage> send(Message message) async {
+    calls++;
+    throw MailException('server is down');
+  }
+}
+
+late BrokenTransport brokenMailer;
+
+Message _message() => Message(
+  to: <Address>[const Address('ann@example.com', 'Ann')],
+  bcc: <Address>[const Address('boss@example.com')],
+  subject: 'Your invoice',
+  text: 'Hi Ann',
+  html: '<p>Hi Ann</p>',
+  attachments: <Attachment>[
+    Attachment(<int>[0, 1, 2, 255], 'invoice.pdf'),
+  ],
+);
+
+/// take one job of [queue] and run it with an app registry that has no
+/// mail job, like a real app
+Future<JobPayload> _work([String queue = 'default', int tries = 1]) async {
+  ReservedJob? job = await Queue.driver.pop(queue);
+  expect(job, isNotNull, reason: 'no job on $queue');
+  await Worker(
+    driver: Queue.driver,
+    jobs: const JobRegistry(<JobCodec>[]),
+    options: WorkerOptions(tries: tries),
+  ).process(job!);
+  return job.payload;
+}
+
+void main() {
+  setUpAll(setUpTestApp);
+
+  setUp(() {
+    Mail.restore();
+    mainMailer = ArrayTransport();
+    otherMailer = ArrayTransport();
+    brokenMailer = BrokenTransport();
+    Mail.configure(
+      MailConfig(
+        defaultMailer: 'main',
+        from: const Address('app@example.com', 'App'),
+        mailers: <String, Transport>{
+          'main': mainMailer,
+          'other': otherMailer,
+          'broken': brokenMailer,
+        },
+      ),
+    );
+  });
+
+  tearDown(Queue.close);
+
+  test('with the sync driver, Mail.queue sends at once', () async {
+    Queue.configure(QueueConfig());
+    await Mail.mailer('other').queue(_message());
+    expect(otherMailer.messages.single.subject, 'Your invoice');
+    expect(mainMailer.sent, isEmpty);
+  });
+
+  for (TestStore store in testStores) {
+    group(store.name, skip: store.skip, () {
+      setUp(() async {
+        await store.reset();
+        Queue.configure(store.config());
+      });
+
+      test('Mail.queue saves the mail, the worker sends it', () async {
+        String id = await Mail.queue(_message());
+        expect(mainMailer.sent, isEmpty);
+        JobPayload payload = await _work();
+        expect(payload.id, id);
+        expect(payload.job, 'dox_queue:SendQueuedMail');
+        Message sent = mainMailer.messages.single;
+        expect(sent.from, const Address('app@example.com', 'App'));
+        expect(sent.to, <Address>[const Address('ann@example.com', 'Ann')]);
+        expect(sent.bcc, <Address>[const Address('boss@example.com')]);
+        expect(sent.html, '<p>Hi Ann</p>');
+        expect(sent.attachments.single.bytes, <int>[0, 1, 2, 255]);
+        expect(sent.attachments.single.contentType, 'application/pdf');
+        expect(await Queue.driver.failed(), isEmpty);
+      });
+
+      test('the worker sends with the same mailer', () async {
+        await Mail.mailer('other').queue(_message(), queue: 'emails');
+        await _work('emails');
+        expect(otherMailer.sent, hasLength(1));
+        expect(mainMailer.sent, isEmpty);
+      });
+
+      test('Mail.later waits for the delay', () async {
+        await Mail.later(const Duration(minutes: 5), _message());
+        expect(await Queue.driver.pop('default'), isNull);
+        expect(mainMailer.sent, isEmpty);
+      });
+
+      test('a failing transport is retried, then the job fails', () async {
+        await Mail.mailer('broken').queue(_message());
+        await _work('default', 2);
+        expect(brokenMailer.calls, 1);
+        expect(await Queue.driver.failed(), isEmpty);
+        await _work('default', 2);
+        expect(brokenMailer.calls, 2);
+        List<FailedJob> failed = await Queue.driver.failed();
+        expect(failed.single.job, 'dox_queue:SendQueuedMail');
+        expect(failed.single.error, contains('server is down'));
+        expect(await Queue.driver.pop('default'), isNull);
+      });
+
+      test('encrypt: true saves the mail encrypted', () async {
+        await Mail.queue(_message(), encrypt: true);
+        JobPayload payload = await _work();
+        expect(payload.job, 'dox_queue:SendEncryptedQueuedMail');
+        expect(payload.data, isNot(contains('ann@example.com')));
+        expect(payload.data, isNot(contains('Your invoice')));
+        expect(payload.data, isNot(contains('Hi Ann')));
+        expect(mainMailer.messages.single.subject, 'Your invoice');
+      });
+    });
+  }
+}