Merge pull request #14580 from misskey-dev/develop

Release: 2024.9.0

1 files changed, 44 insertions, 0 deletions

diff --git a/packages/backend/src/misc/collapsed-queue.ts b/packages/backend/src/misc/collapsed-queue.ts
new file mode 100644
index 0000000000..5bc20a78ae
--- /dev/null
+++ b/packages/backend/src/misc/collapsed-queue.ts
@@ -0,0 +1,44 @@
+/*
+ * SPDX-FileCopyrightText: syuilo and misskey-project
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+type Job<V> = {
+	value: V;
+	timer: NodeJS.Timeout;
+};
+
+// TODO: redis使えるようにする
+export class CollapsedQueue<K, V> {
+	private jobs: Map<K, Job<V>> = new Map();
+
+	constructor(
+		private timeout: number,
+		private collapse: (oldValue: V, newValue: V) => V,
+		private perform: (key: K, value: V) => Promise<void>,
+	) {}
+
+	enqueue(key: K, value: V) {
+		if (this.jobs.has(key)) {
+			const old = this.jobs.get(key)!;
+			const merged = this.collapse(old.value, value);
+			this.jobs.set(key, { ...old, value: merged });
+		} else {
+			const timer = setTimeout(() => {
+				const job = this.jobs.get(key)!;
+				this.jobs.delete(key);
+				this.perform(key, job.value);
+			}, this.timeout);
+			this.jobs.set(key, { value, timer });
+		}
+	}
+
+	async performAllNow() {
+		const entries = [...this.jobs.entries()];
+		this.jobs.clear();
+		for (const [_key, job] of entries) {
+			clearTimeout(job.timer);
+		}
+		await Promise.allSettled(entries.map(([key, job]) => this.perform(key, job.value)));
+	}
+}