From 05b2f244dd679e79a8b9aafc241e32e02b0d3b63 Mon Sep 17 00:00:00 2001 From: necessarylion Date: Sun, 1 Nov 2026 22:47:29 +0700 Subject: [PATCH 1/3] feat: built-in job for Mail.queue of dox_mailer Mail.queue() of dox_mailer needs a job that sends the saved email. dox_queue now has it: SendQueuedMail, and SendEncryptedQueuedMail for encrypt: true. Queue.configure() gives it to Mail.useQueue(). The builder only finds jobs in the app lib folder, so a package job is not in jobs.g.dart. JobRegistry now also finds the built-in jobs, after the jobs of the app. dox_queue now depends on dox_mailer (light: http and crypto). So dox_mailer must be released first. --- CHANGELOG.md | 2 + README.md | 11 ++++ lib/src/job_registry.dart | 13 ++++- lib/src/queue.dart | 5 ++ lib/src/send_queued_mail.dart | 94 +++++++++++++++++++++++++++++++ pubspec.yaml | 1 + test/queue_mail_test.dart | 103 ++++++++++++++++++++++++++++++++++ 7 files changed, 227 insertions(+), 2 deletions(-) create mode 100644 lib/src/send_queued_mail.dart create mode 100644 test/queue_mail_test.dart diff --git a/CHANGELOG.md b/CHANGELOG.md index 3833680..8877e83 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,7 @@ ## 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. It must be released before dox_queue. - 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. diff --git a/README.md b/README.md index 68822a8..433adc3 100644 --- a/README.md +++ b/README.md @@ -264,6 +264,17 @@ 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. + ## Run the worker ```bash diff --git a/lib/src/job_registry.dart b/lib/src/job_registry.dart index a500ce6..b4b8000 100644 --- a/lib/src/job_registry.dart +++ b/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 Function(DoxJob job); @@ -24,10 +25,18 @@ class JobRegistry { final List 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 builtIn = [ + 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; @@ -35,7 +44,7 @@ class JobRegistry { /// 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; diff --git a/lib/src/queue.dart b/lib/src/queue.dart index ab7202b..fdb03ce 100644 --- a/lib/src/queue.dart +++ b/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] diff --git a/lib/src/send_queued_mail.dart b/lib/src/send_queued_mail.dart new file mode 100644 index 0000000..31294cd --- /dev/null +++ b/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 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 handle() async { + await Mail.mailer(mailer).send(Message.fromJson(message)); + } + + static Map encode(DoxJob job) { + SendQueuedMail value = job as SendQueuedMail; + return {'mailer': value.mailer, 'message': value.message}; + } + + static DoxJob _decode(Map data) { + return SendQueuedMail( + mailer: data['mailer'] as String, + message: Map.from( + data['message'] as Map, + ), + ); + } +} + +/// [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 data) { + return SendEncryptedQueuedMail( + mailer: data['mailer'] as String, + message: Map.from( + data['message'] as Map, + ), + ); + } +} + +/// Puts the mails of `Mail.queue()` on this queue. `Queue.configure()` +/// sets it with `Mail.useQueue()`. +class QueueMailQueue implements MailQueue { + const QueueMailQueue(); + + @override + Future push( + String mailer, + Message message, { + Duration? delay, + String? queue, + bool encrypt = false, + }) { + Map data = message.toJson(); + DoxJob job = encrypt + ? SendEncryptedQueuedMail(mailer: mailer, message: data) + : SendQueuedMail(mailer: mailer, message: data); + return Queue.dispatch(job, delay: delay, queue: queue); + } +} diff --git a/pubspec.yaml b/pubspec.yaml index 217a7ce..9d2f776 100644 --- a/pubspec.yaml +++ b/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 diff --git a/test/queue_mail_test.dart b/test/queue_mail_test.dart new file mode 100644 index 0000000..ff66f43 --- /dev/null +++ b/test/queue_mail_test.dart @@ -0,0 +1,103 @@ +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; + +Message _message() => Message( + to:
[const Address('ann@example.com', 'Ann')], + bcc:
[const Address('boss@example.com')], + subject: 'Your invoice', + text: 'Hi Ann', + html: '

Hi Ann

', + attachments: [ + Attachment([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 _work([String queue = 'default']) async { + ReservedJob? job = await Queue.driver.pop(queue); + expect(job, isNotNull, reason: 'no job on $queue'); + await Worker( + driver: Queue.driver, + jobs: const JobRegistry([]), + ).process(job!); + return job.payload; +} + +void main() { + setUpAll(setUpTestApp); + + setUp(() { + Mail.restore(); + mainMailer = ArrayTransport(); + otherMailer = ArrayTransport(); + Mail.configure( + MailConfig( + defaultMailer: 'main', + from: const Address('app@example.com', 'App'), + mailers: {'main': mainMailer, 'other': otherMailer}, + ), + ); + }); + + 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,
[const Address('ann@example.com', 'Ann')]); + expect(sent.bcc,
[const Address('boss@example.com')]); + expect(sent.html, '

Hi Ann

'); + expect(sent.attachments.single.bytes, [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('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(mainMailer.messages.single.subject, 'Your invoice'); + }); + }); + } +} From b2f8ee7ced04c889f33114f887a6fb637e694e4c Mon Sep 17 00:00:00 2001 From: necessarylion Date: Sun, 1 Nov 2026 22:53:30 +0700 Subject: [PATCH 2/3] fix: review nits for the built-in mail job The changelog no longer has the release note for maintainers. It now says to deploy the worker with the web app, because an old worker can not read the new mail job. The README says it too. The job builder now refuses an app job name that starts with dox_queue:, so it can not clash with a built-in job. New test: a failing transport goes through the normal tries and failed jobs. The encrypt test now also checks that the subject and the body are hidden. --- CHANGELOG.md | 3 +- README.md | 2 ++ lib/src/builder/job_registry_builder.dart | 6 ++++ test/builder_test.dart | 12 +++++++ test/queue_mail_test.dart | 38 +++++++++++++++++++++-- 5 files changed, 58 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8877e83..8c466c5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,7 +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. It must be released before dox_queue. +- 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. diff --git a/README.md b/README.md index 433adc3..8d6dcaa 100644 --- a/README.md +++ b/README.md @@ -275,6 +275,8 @@ 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 diff --git a/lib/src/builder/job_registry_builder.dart b/lib/src/builder/job_registry_builder.dart index 6594e48..ceff4a6 100644 --- a/lib/src/builder/job_registry_builder.dart +++ b/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; diff --git a/test/builder_test.dart b/test/builder_test.dart index c704181..4a73480 100644 --- a/test/builder_test.dart +++ b/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 handle() async {} +} +'''); + expect(error, contains('kept for the built-in jobs')); + }); } diff --git a/test/queue_mail_test.dart b/test/queue_mail_test.dart index ff66f43..b2f2f8d 100644 --- a/test/queue_mail_test.dart +++ b/test/queue_mail_test.dart @@ -7,6 +7,19 @@ import 'support.dart'; late ArrayTransport mainMailer; late ArrayTransport otherMailer; +/// a mail server that is down +class BrokenTransport extends Transport { + int calls = 0; + + @override + Future send(Message message) async { + calls++; + throw MailException('server is down'); + } +} + +late BrokenTransport brokenMailer; + Message _message() => Message( to:
[const Address('ann@example.com', 'Ann')], bcc:
[const Address('boss@example.com')], @@ -20,12 +33,13 @@ Message _message() => Message( /// take one job of [queue] and run it with an app registry that has no /// mail job, like a real app -Future _work([String queue = 'default']) async { +Future _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([]), + options: WorkerOptions(tries: tries), ).process(job!); return job.payload; } @@ -37,11 +51,16 @@ void main() { Mail.restore(); mainMailer = ArrayTransport(); otherMailer = ArrayTransport(); + brokenMailer = BrokenTransport(); Mail.configure( MailConfig( defaultMailer: 'main', from: const Address('app@example.com', 'App'), - mailers: {'main': mainMailer, 'other': otherMailer}, + mailers: { + 'main': mainMailer, + 'other': otherMailer, + 'broken': brokenMailer, + }, ), ); }); @@ -91,11 +110,26 @@ void main() { 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 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'); }); }); From 885681e1b79db587f9f74713eb48f741f2341729 Mon Sep 17 00:00:00 2001 From: necessarylion Date: Mon, 2 Nov 2026 06:14:11 +0700 Subject: [PATCH 3/3] ci: test against the newest dox-mailer dox_queue now needs Mail.queue of dox_mailer. That code is on dox-mailer main, but not on pub.dev yet. So CI took dox_mailer from pub.dev and failed. ci/use-siblings.sh now clones dox-mailer too. --- ci/use-siblings.sh | 1 + 1 file changed, 1 insertion(+) diff --git a/ci/use-siblings.sh b/ci/use-siblings.sh index f8d8b17..736a28c 100755 --- a/ci/use-siblings.sh +++ b/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