diff --git a/CHANGELOG.md b/CHANGELOG.md index 3833680..8c466c5 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/README.md b/README.md index 68822a8..8d6dcaa 100644 --- a/README.md +++ b/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 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 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/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/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 new file mode 100644 index 0000000..b2f2f8d --- /dev/null +++ b/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 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')], + 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', 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; +} + +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: { + '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,
[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('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'); + }); + }); + } +}