fix(server): mail test & retry (#15044)

#### PR Dependency Tree


* **PR #15044** 👈

This tree was auto-generated by
[Charcoal](https://github.com/danerwilliams/charcoal)

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **Bug Fixes**
* Stop sending notifications to disabled users; skip member invites when
workspace names contain URLs/domains
* Improve mail retry handling (per-recipient exhaustion, expiry, and
cache cleanup)
  * Make many email headers/lead lines more generic and consistent
  * Fail-safe workspace content parsing to avoid crashes

* **New Features**
* 24-hour signup protection for sharing, invites, and invite-link
creation
  * Job-queue: remove jobs by payload predicate

* **Tests**
* Expanded tests for mail jobs, SMTP hostname handling, payment
checkout, job-queue removal, and abuse-detection utilities
  * Updated test fixtures to set createdAt timestamps for new users

* **Chores**
  * Added required name input for test-email mutation
  * Database flush retry with deadlock detection/backoff

<!-- review_stack_entry_start -->

[![Review Change
Stack](https://storage.googleapis.com/coderabbit_public_assets/review-stack-in-coderabbit-ui.svg)](https://app.coderabbit.ai/change-stack/toeverything/AFFiNE/pull/15044?utm_source=github_walkthrough&utm_medium=github&utm_campaign=change_stack)

<!-- review_stack_entry_end -->
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

packages/backend/server/src/base/job/queue/__tests__/queue.spec.ts

@@ -117,6 +117,24 @@ test('should remove job from queue', async t => {
   t.is(nullData, undefined);
   t.is(nullJob, undefined);
 });
+
+test('should remove jobs by payload predicate', async t => {
+  const keep = await queue.add('nightly.__test__job', { name: 'keep' });
+  const remove = await queue.add('nightly.__test__job', { name: 'remove' });
+  const other = await queue.add('nightly.__test__job2', { name: 'remove' });
+  const getJobs = Sinon.spy(bullmq, 'getJobs');
+
+  const removed = await queue.removeWhere(
+    'nightly.__test__job',
+    job => job.name === 'remove'
+  );
+
+  t.deepEqual(getJobs.firstCall.args.slice(0, 3), [['waiting'], 0, 99]);
+  t.deepEqual(removed, [{ name: 'remove' }]);
+  t.truthy(await queue.get(keep.id!, 'nightly.__test__job'));
+  t.is(await queue.get(remove.id!, 'nightly.__test__job'), undefined);
+  t.truthy(await queue.get(other.id!, 'nightly.__test__job2'));
+});
 // #endregion
 
 // #region executor

packages/backend/server/src/base/job/queue/queue.ts

@@ -12,6 +12,15 @@ interface JobData<T extends JobName> {
   payload: Jobs[T];
 }
 
+const removableJobStates = [
+  'waiting',
+  'delayed',
+  'prioritized',
+  'paused',
+  'waiting-children',
+] as const;
+const removeWhereBatchSize = 100;
+
 @Injectable()
 export class JobQueue {
   private readonly logger = new Logger(JobQueue.name);
@@ -55,6 +64,59 @@ export class JobQueue {
     return undefined;
   }
 
+  async removeWhere<T extends JobName>(
+    jobName: T,
+    predicate: (payload: Jobs[T]) => boolean | Promise<boolean>
+  ): Promise<Jobs[T][]> {
+    const ns = namespace(jobName);
+    const queue = this.getQueue(ns);
+    const removed: Jobs[T][] = [];
+
+    for (const state of removableJobStates) {
+      let start = 0;
+      let removedFromBatch = false;
+      let hasMoreJobs = true;
+
+      while (hasMoreJobs) {
+        removedFromBatch = false;
+        const jobs = (await queue.getJobs(
+          [state],
+          start,
+          start + removeWhereBatchSize - 1
+        )) as Job<JobData<T>>[];
+
+        if (!jobs.length) {
+          hasMoreJobs = false;
+          break;
+        }
+
+        for (const job of jobs) {
+          if (job.name !== jobName) {
+            continue;
+          }
+
+          const payload = job.data.payload;
+          if (!(await predicate(payload))) {
+            continue;
+          }
+
+          await job.remove();
+          this.logger.log(
+            `Job ${jobName}(id=${job.id}) removed from queue ${ns}`
+          );
+          removed.push(payload);
+          removedFromBatch = true;
+        }
+
+        if (!removedFromBatch) {
+          start += removeWhereBatchSize;
+        }
+      }
+    }
+
+    return removed;
+  }
+
   async get<T extends JobName>(jobId: string, jobName: T) {
     const ns = namespace(jobName);
     const queue = this.getQueue(ns);