improve api queue management with lifo and priority bumping

- convert queue from fifo to lifo (process newest requests first)
- add 10 element queue limit, drop oldest when full
- add 1 minute age limit, skip stale requests
- add priority bumping: if user re-queued, remove old request and process new one first
- clear queue on navigation to prevent stale requests
- replace setInterval with promise.race for navigation detection

Author: chefnaphtha

package.json

@@ -1,6 +1,6 @@
 {
 	"name": "xblockorigin",
-	"version": "23.0.0",
+	"version": "24.0.0",
 	"type": "module",
 	"scripts": {
 		"postinstall": "bun run setup-hooks",

packages/extension/src/Content/orchestrator.ts

@@ -49,7 +49,7 @@ function showToast(message: string) {
 async function processUser(username: string, tweetElement?: Element) {
 	// skip if already processing this user
 	if (inFlightUsers.has(username)) {
-		console.log(`[xBlockOrigin] Skipping @${username} - already processing`)
+		console.log(`[xBlockOrigin] Skipping @${username} - already in queue`)
 		return
 	}
 	inFlightUsers.add(username)
@@ -67,7 +67,10 @@ async function processUser(username: string, tweetElement?: Element) {
 		console.log(
 			`[xBlockOrigin] Fetching user data for @${username} (cache miss)`
 		)
-		const userData = await apiQueue.enqueue(() => getUserData(username))
+		const userData = await apiQueue.enqueue(
+			() => getUserData(username),
+			username
+		)
 
 		if (!userData) {
 			console.error(
@@ -102,7 +105,10 @@ async function processUser(username: string, tweetElement?: Element) {
 			console.log(
 				`[xBlockOrigin] Following status cache miss for @${username}, fetching user data`
 			)
-			const userData = await apiQueue.enqueue(() => getUserData(username))
+			const userData = await apiQueue.enqueue(
+				() => getUserData(username),
+				username
+			)
 
 			if (!userData) {
 				console.error(
@@ -151,7 +157,7 @@ async function processUser(username: string, tweetElement?: Element) {
 
 		// fetch country for new users
 		console.log(`[xBlockOrigin] Fetching country for @${username}`)
-		country = await apiQueue.enqueue(() => getCountry(username))
+		country = await apiQueue.enqueue(() => getCountry(username), username)
 
 		if (!country) {
 			console.log(
@@ -196,7 +202,7 @@ async function processUser(username: string, tweetElement?: Element) {
 		console.log(
 			`[xBlockOrigin] Attempting to mute @${username} (${userId}) from ${country}...`
 		)
-		const success = await apiQueue.enqueue(() => muteUser(userId))
+		const success = await apiQueue.enqueue(() => muteUser(userId), username)
 
 		if (!success) {
 			console.error(`[xBlockOrigin] Failed to mute @${username}`)
@@ -252,64 +258,85 @@ function getCurrentPage(): string {
 	return 'unknown'
 }
 
+function waitForNavigation(currentUrl: string): Promise<string> {
+	// race between popstate event and polling
+	const popstatePromise = new Promise<string>((resolve) => {
+		const handler = () => {
+			if (window.location.href !== currentUrl) {
+				resolve(window.location.href)
+			}
+		}
+		window.addEventListener('popstate', handler, { once: true })
+	})
+
+	const pollingPromise = new Promise<string>((resolve) => {
+		const checkUrl = () => {
+			if (window.location.href !== currentUrl) {
+				resolve(window.location.href)
+			} else {
+				setTimeout(checkUrl, 100)
+			}
+		}
+		checkUrl()
+	})
+
+	return Promise.race([popstatePromise, pollingPromise])
+}
+
 export function startOrchestrator() {
 	const cleanupFns: Array<() => void> = []
+	let running = true
 
 	const handleUser = (username: string, tweetElement?: Element) => {
 		processUser(username, tweetElement)
 	}
 
-	const currentPage = getCurrentPage()
-
-	switch (currentPage) {
-		case 'timeline':
-			cleanupFns.push(scanTimeline(handleUser))
-			break
-		case 'search':
-			cleanupFns.push(scanSearch(handleUser))
-			break
-		case 'notifications':
-			cleanupFns.push(scanReplies(handleUser))
-			break
-		case 'status':
-			cleanupFns.push(scanStatus(handleUser))
-			break
-		case 'profile':
-			cleanupFns.push(scanProfile(handleUser))
-			break
+	const startScanners = (page: string) => {
+		switch (page) {
+			case 'timeline':
+				cleanupFns.push(scanTimeline(handleUser))
+				break
+			case 'search':
+				cleanupFns.push(scanSearch(handleUser))
+				break
+			case 'notifications':
+				cleanupFns.push(scanReplies(handleUser))
+				break
+			case 'status':
+				cleanupFns.push(scanStatus(handleUser))
+				break
+			case 'profile':
+				cleanupFns.push(scanProfile(handleUser))
+				break
+		}
 	}
 
-	let lastUrl = window.location.href
-	const urlWatcher = setInterval(() => {
-		if (window.location.href !== lastUrl) {
-			lastUrl = window.location.href
+	const handleNavigation = async () => {
+		let currentUrl = window.location.href
+		startScanners(getCurrentPage())
+
+		while (running) {
+			const newUrl = await waitForNavigation(currentUrl)
+			if (!running) break
+
+			currentUrl = newUrl
 
+			// clear pending API requests on navigation
+			apiQueue.clear()
+
+			// cleanup old scanners
 			cleanupFns.forEach((fn) => fn())
 			cleanupFns.length = 0
 
-			const newPage = getCurrentPage()
-			switch (newPage) {
-				case 'timeline':
-					cleanupFns.push(scanTimeline(handleUser))
-					break
-				case 'search':
-					cleanupFns.push(scanSearch(handleUser))
-					break
-				case 'notifications':
-					cleanupFns.push(scanReplies(handleUser))
-					break
-				case 'status':
-					cleanupFns.push(scanStatus(handleUser))
-					break
-				case 'profile':
-					cleanupFns.push(scanProfile(handleUser))
-					break
-			}
+			// start new scanners
+			startScanners(getCurrentPage())
 		}
-	}, 1000)
+	}
+
+	handleNavigation()
 
 	return () => {
-		clearInterval(urlWatcher)
+		running = false
 		cleanupFns.forEach((fn) => fn())
 	}
 }

packages/extension/src/Utils/rateLimit.ts

@@ -2,8 +2,13 @@ type QueuedTask = {
 	fn: () => Promise<unknown>
 	resolve: (value: unknown) => void
 	reject: (error: unknown) => void
+	addedAt: number
+	key?: string
 }
 
+const MAX_AGE_MS = 60 * 1000 // 1 minute
+const MAX_QUEUE_SIZE = 10
+
 export function createQueue() {
 	const queue: QueuedTask[] = []
 	let isProcessing = false
@@ -16,9 +21,23 @@ export function createQueue() {
 		isProcessing = true
 
 		while (queue.length > 0) {
-			const task = queue.shift()
+			const task = queue.pop()
 			if (!task) break
 
+			// skip tasks older than 1 minute
+			const age = Date.now() - task.addedAt
+			if (age > MAX_AGE_MS) {
+				console.warn(
+					`[xBlockOrigin] Skipping stale API request (age: ${Math.round(age / 1000)}s)`
+				)
+				task.reject(
+					new Error(
+						`Request timed out after ${MAX_AGE_MS}ms in queue`
+					)
+				)
+				continue
+			}
+
 			try {
 				const result = await task.fn()
 				task.resolve(result)
@@ -31,19 +50,67 @@ export function createQueue() {
 	}
 
 	return {
-		enqueue<T>(fn: () => Promise<T>): Promise<T> {
+		enqueue<T>(fn: () => Promise<T>, key?: string): Promise<T> {
 			return new Promise<T>((resolve, reject) => {
-				queue.push({
+				// if key is provided, check if task already exists and remove it
+				if (key) {
+					const existingIndex = queue.findIndex(
+						(task) => task.key === key
+					)
+					if (existingIndex !== -1) {
+						const removed = queue.splice(existingIndex, 1)[0]
+						if (removed) {
+							console.log(
+								`[xBlockOrigin] Bumping priority for ${key} (was at position ${existingIndex})`
+							)
+							removed.reject(
+								new Error('Task re-queued with higher priority')
+							)
+						}
+					}
+				}
+
+				// remove oldest item if queue is full
+				if (queue.length >= MAX_QUEUE_SIZE) {
+					const removed = queue.shift()
+					if (removed) {
+						console.warn(
+							`[xBlockOrigin] Queue full, dropping oldest request (age: ${Math.round((Date.now() - removed.addedAt) / 1000)}s)`
+						)
+						removed.reject(new Error('Queue full, request dropped'))
+					}
+				}
+
+				const task: QueuedTask = {
 					fn,
 					resolve: (value) => resolve(value as T),
-					reject
-				})
+					reject,
+					addedAt: Date.now()
+				}
+				if (key) {
+					task.key = key
+				}
+				queue.push(task)
 				processQueue()
 			})
 		},
 
 		getQueueLength(): number {
 			return queue.length
+		},
+
+		clear(): void {
+			const count = queue.length
+			if (count > 0) {
+				console.log(
+					`[xBlockOrigin] Clearing ${count} pending API requests from queue`
+				)
+				// reject all pending tasks
+				for (const task of queue) {
+					task.reject(new Error('Queue cleared due to navigation'))
+				}
+				queue.length = 0
+			}
 		}
 	}
 }