Implement Cloudflare Queues for GitHub API rate limiting

- Updated README.md to include Cloudflare Queues for managing GitHub API requests.
- Modified wrangler.toml to configure the GitHub task queue.
- Enhanced handlers to send messages to the queue for fetching repository issues, contributors, and pull requests.
- Added queue processing logic in index.ts to handle queued tasks.
- Introduced new types in types.ts for GitHub task messages.

These changes improve the application's ability to handle GitHub API interactions efficiently while respecting rate limits.

Author: shyim

README.md

@@ -10,6 +10,7 @@ This project is a port of the Friends of Shopware API to Cloudflare Workers, usi
 - Packagist packages information
 - GitHub webhook for issue updates
 - Data cached in Cloudflare KV storage
+- Uses Cloudflare Queues for rate-limited GitHub API operations
 
 ## Setup
 
@@ -19,14 +20,24 @@ This project is a port of the Friends of Shopware API to Cloudflare Workers, usi
 npm install
 ```
 
-2. Configure your KV namespace:
+2. Configure your KV namespace and Queue:
 
-Create a KV namespace in your Cloudflare Workers dashboard and update the `wrangler.toml` file with your KV namespace IDs.
+Create a KV namespace and a Queue in your Cloudflare Workers dashboard and update the `wrangler.toml` file with your IDs.
 
 ```toml
 kv_namespaces = [
   { binding = "STORAGE", id = "your-kv-namespace-id", preview_id = "your-preview-kv-namespace-id" }
 ]
+
+[[queues.producers]]
+queue = "github-tasks"
+binding = "GITHUB_QUEUE"
+
+[[queues.consumers]]
+queue = "github-tasks"
+max_batch_size = 10
+max_batch_timeout = 30
+max_retries = 3
 ```
 
 3. Set your GitHub token:
@@ -63,11 +74,26 @@ npm run deploy
 - `GET /v2/packagist/packages` - List all packages
 - `POST /webhook/issue` - GitHub webhook for issue updates
 
-## Scheduled Tasks
+## Queue Implementation
+
+The project uses Cloudflare Queues to manage GitHub API requests. This helps to:
+
+1. Respect GitHub API rate limits
+2. Make API interactions more resilient
+3. Process tasks asynchronously
+
+Tasks are dispatched to the queues during cron jobs:
 
 - Hourly: Refresh GitHub stats and Packagist stats
 - Every 5 minutes: Refresh repository issues
 
+## Scheduled Tasks
+
+The application uses Cloudflare Cron Triggers to perform these tasks:
+
+- `0 * * * *` (Hourly): Refresh GitHub stats and Packagist stats
+- `*/5 * * * *` (Every 5 minutes): Refresh repository issues
+
 ## License
 
 MIT 

src/handlers.ts

@@ -2,7 +2,7 @@ import { Context } from 'hono';
 import { GitHubService } from './services/github';
 import { PackagistService } from './services/packagist';
 import { CacheService } from './services/cache';
-import { AppContext, Env, GithubWebhook } from './types';
+import { AppContext, Env, GithubWebhook, GitHubTaskMessage } from './types';
 
 // Main Routes Handler
 export const listRepositories = async (c: Context<{ Bindings: Env }>) => {
@@ -40,19 +40,16 @@ export const githubIssueWebhook = async (c: Context<{ Bindings: Env }>) => {
     const payload: GithubWebhook = await c.req.json();
     const repoName = payload.repository.name;
     const ownerLogin = payload.repository.owner.login;
+
+    // Send a message to the queue to update issues
+    await c.env.GITHUB_QUEUE.send({
+      type: 'fetch-repository-issues',
+      owner: ownerLogin,
+      repo: repoName,
+      timestamp: Date.now()
+    } as GitHubTaskMessage);
 
-    const appContext: AppContext = {
-      env: c.env,
-      octokit: c.env.octokit!
-    };
-    
-    const githubService = new GitHubService(appContext);
-    const cacheService = new CacheService(c.env);
-    
-    const issues = await githubService.getAllIssues(ownerLogin, repoName);
-    await cacheService.updateIssuesForRepository(repoName, issues);
-    
-    return c.text(`Updated issues for ${repoName}`, 200);
+    return c.text(`Queued update for issues of ${repoName}`, 200);
   } catch (error) {
     console.error('Error processing webhook:', error);
     return c.text('Error processing webhook', 500);
@@ -61,7 +58,7 @@ export const githubIssueWebhook = async (c: Context<{ Bindings: Env }>) => {
 
 // Background refreshers
 export const refreshGithubStats = async (context: AppContext) => {
-  console.log('Refreshing Github Stats');
+  console.log('Fetching repositories and queueing GitHub stats tasks');
   const githubService = new GitHubService(context);
   const cacheService = new CacheService(context.env);
 
@@ -70,27 +67,58 @@ export const refreshGithubStats = async (context: AppContext) => {
   const repos = await githubService.getAllRepos(orgName);
   await cacheService.setRepositories(repos);
 
-  // Get and cache contributors
-  const contributors = await githubService.getUserContributions(repos);
-  await cacheService.setContributors(contributors);
+  // Queue contributor and PR tasks for each repository
+  for (const repo of repos) {
+    // Queue contributors fetch
+    await context.env.GITHUB_QUEUE.send({
+      type: 'fetch-repository-contributors',
+      owner: repo.owner.login,
+      repo: repo.name,
+      timestamp: Date.now()
+    } as GitHubTaskMessage);
+    
+    // Queue pull requests fetch
+    await context.env.GITHUB_QUEUE.send({
+      type: 'fetch-repository-pull-requests',
+      owner: repo.owner.login,
+      repo: repo.name,
+      timestamp: Date.now()
+    } as GitHubTaskMessage);
+  }
 
-  console.log('Refreshed Github Stats');
+  // Queue a task to process all contributor data after the individual tasks
+  // We'll send this separately without a delay since Cloudflare Workers queues
+  // will process messages in roughly the order they were received
+  await context.env.GITHUB_QUEUE.send({
+    type: 'process-contributors',
+    owner: orgName,
+    timestamp: Date.now(),
+    metadata: {
+      // Add a higher priority flag that we can check in the consumer
+      highPriority: false
+    }
+  } as GitHubTaskMessage);
+  
+  console.log('Queued GitHub stats tasks for all repositories');
 };
 
 export const refreshRepositoryIssues = async (context: AppContext) => {
-  console.log('Refreshing Repository Issues');
-  const githubService = new GitHubService(context);
+  console.log('Queueing Repository Issues refresh tasks');
   const cacheService = new CacheService(context.env);
 
   const repos = await cacheService.getRepositories();
-  const issues: Record<string, any[]> = {};
 
+  // Queue issue fetching for each repository
   for (const repo of repos) {
-    issues[repo.name] = await githubService.getAllIssues(repo.owner.login, repo.name);
+    await context.env.GITHUB_QUEUE.send({
+      type: 'fetch-repository-issues',
+      owner: repo.owner.login,
+      repo: repo.name,
+      timestamp: Date.now()
+    } as GitHubTaskMessage);
   }
 
-  await cacheService.setIssues(issues);
-  console.log('Refreshed Repository Issues');
+  console.log('Queued Repository Issues tasks');
 };
 
 export const refreshPackagistStats = async (context: AppContext) => {

src/index.ts

@@ -1,7 +1,7 @@
 import { Hono } from 'hono';
 import { cors } from 'hono/cors';
 import { Octokit } from '@octokit/rest';
-import { Env, AppContext } from './types';
+import { Env, AppContext, GitHubTaskMessage, MessageBatch } from './types';
 import {
   listRepositories,
   listContributors,
@@ -12,6 +12,7 @@ import {
   refreshRepositoryIssues,
   refreshPackagistStats
 } from './handlers';
+import { processGitHubTasks } from './queue';
 
 // Define app with bindings
 const app = new Hono<{ Bindings: Env }>();
@@ -48,8 +49,16 @@ interface ScheduledController {
 export default {
   fetch: app.fetch,
 
+  // Queue handler - must be declared as a function, not an object
+  async queue(batch: MessageBatch<GitHubTaskMessage>, env: Env, ctx: ExecutionContext) {
+    // Get the queue name from the batch
+    if (batch.queue === 'github-tasks') {
+      await processGitHubTasks(batch, env);
+    }
+  },
+  
   // Handle scheduled events
-  scheduled: async (controller: ScheduledController, env: Env, ctx: ExecutionContext) => {
+  async scheduled(controller: ScheduledController, env: Env, ctx: ExecutionContext) {
     // Create context for scheduled handlers
     const context: AppContext = {
       env,
@@ -60,13 +69,14 @@ export default {
 
     // Determine which refresh operation to perform based on cron schedule
     if (controller.cron === '0 * * * *') {
-      console.log('Refreshing Github Stats');
+      console.log('Starting Github Stats hourly job');
       // Hourly jobs
-      ctx.waitUntil(refreshGithubStats(context));
-      ctx.waitUntil(refreshPackagistStats(context));
+      await refreshGithubStats(context);
+      await refreshPackagistStats(context);
     } else if (controller.cron === '*/5 * * * *') {
+      console.log('Starting Repository Issues 5-minute job');
       // Every 5 minutes
-      ctx.waitUntil(refreshRepositoryIssues(context));
+      await refreshRepositoryIssues(context);
     }
   }
 };