Add per-user rate limiting for API requests (#14)

disnet · claude · web-flow · commit 2405bafc84e5 · 2026-01-15T13:57:42.000-08:00
* Implement per-user rate limiting for authenticated API requests

Add D1-based rate limiting with three tiers: expensive endpoints (30/min), standard (100/min), and light (300/min). Includes atomic increment via SQL, automatic window cleanup via cron, and proper 429 response headers.

Co-Authored-By: Claude Haiku 4.5 &lt;noreply@anthropic.com&gt;

* Add D1 migrations to deploy workflow

Run migrations automatically before deploying the worker for both staging and production environments.

Co-Authored-By: Claude Opus 4.5 &lt;noreply@anthropic.com&gt;

* Make migrations idempotent and add continue-on-error

- Add IF NOT EXISTS to CREATE TABLE/INDEX statements in all migrations
- Add continue-on-error to migration steps in deploy workflow
- This handles the case where migrations were previously applied manually

Co-Authored-By: Claude Opus 4.5 &lt;noreply@anthropic.com&gt;

* Add bootstrap migration to mark legacy migrations as applied

The bootstrap runs via d1 execute (not d1 migrations apply) to populate
the d1_migrations table with records for all previously manually-applied
migrations. This allows d1 migrations apply to skip them and only run
new migrations (0015, 0016).

Co-Authored-By: Claude Opus 4.5 &lt;noreply@anthropic.com&gt;

* Add batch feed fetching to reduce API calls on login

The frontend now fetches all ready feeds in a single batch request instead
of making N individual requests. This prevents rate limiting issues when
users have many subscriptions. Pending feeds are still fetched gradually.

Co-Authored-By: Claude Opus 4.5 &lt;noreply@anthropic.com&gt;

---------

Co-authored-by: Claude Haiku 4.5 &lt;noreply@anthropic.com&gt;
diff --git a/.github/workflows/deploy.yml b/.github/workflows/deploy.yml
@@ -70,6 +70,24 @@ jobs:
         working-directory: backend
         run: npm ci
 
+      - name: Bootstrap D1 migrations table (staging)
+        if: needs.changes.outputs.environment == 'staging'
+        uses: cloudflare/wrangler-action@v3
+        with:
+          apiToken: ${{ secrets.CLOUDFLARE_API_TOKEN }}
+          accountId: ${{ secrets.CLOUDFLARE_ACCOUNT_ID }}
+          workingDirectory: backend
+          command: d1 execute skyreader-staging --env=staging --remote --file=migrations/0000_bootstrap.sql
+
+      - name: Run D1 migrations (staging)
+        if: needs.changes.outputs.environment == 'staging'
+        uses: cloudflare/wrangler-action@v3
+        with:
+          apiToken: ${{ secrets.CLOUDFLARE_API_TOKEN }}
+          accountId: ${{ secrets.CLOUDFLARE_ACCOUNT_ID }}
+          workingDirectory: backend
+          command: d1 migrations apply skyreader-staging --env=staging --remote
+
       - name: Deploy Worker (staging)
         if: needs.changes.outputs.environment == 'staging'
         uses: cloudflare/wrangler-action@v3
@@ -79,6 +97,24 @@ jobs:
           workingDirectory: backend
           command: deploy --env=staging
 
+      - name: Bootstrap D1 migrations table (production)
+        if: needs.changes.outputs.environment != 'staging'
+        uses: cloudflare/wrangler-action@v3
+        with:
+          apiToken: ${{ secrets.CLOUDFLARE_API_TOKEN }}
+          accountId: ${{ secrets.CLOUDFLARE_ACCOUNT_ID }}
+          workingDirectory: backend
+          command: d1 execute skyreader --remote --file=migrations/0000_bootstrap.sql
+
+      - name: Run D1 migrations (production)
+        if: needs.changes.outputs.environment != 'staging'
+        uses: cloudflare/wrangler-action@v3
+        with:
+          apiToken: ${{ secrets.CLOUDFLARE_API_TOKEN }}
+          accountId: ${{ secrets.CLOUDFLARE_ACCOUNT_ID }}
+          workingDirectory: backend
+          command: d1 migrations apply skyreader --remote
+
       - name: Deploy Worker (production)
         if: needs.changes.outputs.environment != 'staging'
         uses: cloudflare/wrangler-action@v3
diff --git a/backend/migrations/0000_bootstrap.sql b/backend/migrations/0000_bootstrap.sql
@@ -0,0 +1,26 @@
+-- Bootstrap: Mark all legacy migrations as applied
+-- This handles the case where migrations were applied manually via d1 execute
+-- before switching to d1 migrations apply
+
+-- Create the migrations tracking table if it doesn't exist
+CREATE TABLE IF NOT EXISTS d1_migrations (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    name TEXT UNIQUE NOT NULL,
+    applied_at TEXT NOT NULL DEFAULT (datetime('now'))
+);
+
+-- Insert records for all legacy migrations (ignore if already exists)
+INSERT OR IGNORE INTO d1_migrations (name) VALUES ('0001_initial.sql');
+INSERT OR IGNORE INTO d1_migrations (name) VALUES ('0002_scheduled_feeds.sql');
+INSERT OR IGNORE INTO d1_migrations (name) VALUES ('0003_read_positions_cache.sql');
+INSERT OR IGNORE INTO d1_migrations (name) VALUES ('0004_follows_sync.sql');
+INSERT OR IGNORE INTO d1_migrations (name) VALUES ('0005_share_guid_published.sql');
+INSERT OR IGNORE INTO d1_migrations (name) VALUES ('0006_share_feed_url.sql');
+INSERT OR IGNORE INTO d1_migrations (name) VALUES ('0007_share_content.sql');
+INSERT OR IGNORE INTO d1_migrations (name) VALUES ('0008_sessions.sql');
+INSERT OR IGNORE INTO d1_migrations (name) VALUES ('0009_oauth_state.sql');
+INSERT OR IGNORE INTO d1_migrations (name) VALUES ('0010_feed_cache.sql');
+INSERT OR IGNORE INTO d1_migrations (name) VALUES ('0011_feed_items.sql');
+INSERT OR IGNORE INTO d1_migrations (name) VALUES ('0012_inapp_follows.sql');
+INSERT OR IGNORE INTO d1_migrations (name) VALUES ('0013_share_read_positions_cache.sql');
+INSERT OR IGNORE INTO d1_migrations (name) VALUES ('0014_feed_metadata_image_url.sql');
diff --git a/backend/migrations/0001_initial.sql b/backend/migrations/0001_initial.sql
@@ -1,5 +1,5 @@
 -- Users table - caches AT Protocol user information
-CREATE TABLE users (
+CREATE TABLE IF NOT EXISTS users (
     did TEXT PRIMARY KEY,
     handle TEXT NOT NULL,
     display_name TEXT,
@@ -10,21 +10,21 @@ CREATE TABLE users (
     updated_at INTEGER NOT NULL DEFAULT (unixepoch())
 );
 
-CREATE INDEX idx_users_handle ON users(handle);
+CREATE INDEX IF NOT EXISTS idx_users_handle ON users(handle);
 
 -- Follows cache - which users follow whom (from Bluesky)
-CREATE TABLE follows_cache (
+CREATE TABLE IF NOT EXISTS follows_cache (
     follower_did TEXT NOT NULL,
     following_did TEXT NOT NULL,
     created_at INTEGER NOT NULL DEFAULT (unixepoch()),
     PRIMARY KEY (follower_did, following_did),
     FOREIGN KEY (follower_did) REFERENCES users(did) ON DELETE CASCADE
 );
 
-CREATE INDEX idx_follows_following ON follows_cache(following_did);
+CREATE INDEX IF NOT EXISTS idx_follows_following ON follows_cache(following_did);
 
 -- Aggregated shares from all users
-CREATE TABLE shares (
+CREATE TABLE IF NOT EXISTS shares (
     id INTEGER PRIMARY KEY AUTOINCREMENT,
     author_did TEXT NOT NULL,
     record_uri TEXT UNIQUE NOT NULL,
@@ -41,12 +41,12 @@ CREATE TABLE shares (
     FOREIGN KEY (author_did) REFERENCES users(did) ON DELETE CASCADE
 );
 
-CREATE INDEX idx_shares_author ON shares(author_did);
-CREATE INDEX idx_shares_created ON shares(created_at DESC);
-CREATE INDEX idx_shares_item_url ON shares(item_url);
+CREATE INDEX IF NOT EXISTS idx_shares_author ON shares(author_did);
+CREATE INDEX IF NOT EXISTS idx_shares_created ON shares(created_at DESC);
+CREATE INDEX IF NOT EXISTS idx_shares_item_url ON shares(item_url);
 
 -- Feed metadata cache (for proxy optimization)
-CREATE TABLE feed_metadata (
+CREATE TABLE IF NOT EXISTS feed_metadata (
     feed_url TEXT PRIMARY KEY,
     title TEXT,
     site_url TEXT,
@@ -60,7 +60,7 @@ CREATE TABLE feed_metadata (
 );
 
 -- Sync state tracking (e.g., Jetstream cursor)
-CREATE TABLE sync_state (
+CREATE TABLE IF NOT EXISTS sync_state (
     key TEXT PRIMARY KEY,
     value TEXT NOT NULL,
     updated_at INTEGER NOT NULL DEFAULT (unixepoch())
diff --git a/backend/migrations/0002_scheduled_feeds.sql b/backend/migrations/0002_scheduled_feeds.sql
@@ -7,7 +7,7 @@ CREATE INDEX IF NOT EXISTS idx_users_last_active ON users(last_active_at);
 
 -- Cache of user subscriptions (synced from AT Protocol)
 -- Used by scheduled feed fetcher to know which feeds to refresh
-CREATE TABLE subscriptions_cache (
+CREATE TABLE IF NOT EXISTS subscriptions_cache (
     id INTEGER PRIMARY KEY AUTOINCREMENT,
     user_did TEXT NOT NULL,
     record_uri TEXT UNIQUE NOT NULL,
@@ -17,8 +17,8 @@ CREATE TABLE subscriptions_cache (
     FOREIGN KEY (user_did) REFERENCES users(did) ON DELETE CASCADE
 );
 
-CREATE INDEX idx_subscriptions_cache_user ON subscriptions_cache(user_did);
-CREATE INDEX idx_subscriptions_cache_feed_url ON subscriptions_cache(feed_url);
+CREATE INDEX IF NOT EXISTS idx_subscriptions_cache_user ON subscriptions_cache(user_did);
+CREATE INDEX IF NOT EXISTS idx_subscriptions_cache_feed_url ON subscriptions_cache(feed_url);
 
 -- Add scheduled fetch tracking to feed_metadata
 ALTER TABLE feed_metadata ADD COLUMN last_scheduled_fetch_at INTEGER;
diff --git a/backend/migrations/0003_read_positions_cache.sql b/backend/migrations/0003_read_positions_cache.sql
@@ -1,6 +1,6 @@
 -- Cache of user read positions (synced from AT Protocol)
 -- Used to dedupe before syncing to PDS
-CREATE TABLE read_positions_cache (
+CREATE TABLE IF NOT EXISTS read_positions_cache (
     id INTEGER PRIMARY KEY AUTOINCREMENT,
     user_did TEXT NOT NULL,
     rkey TEXT NOT NULL,
@@ -16,6 +16,6 @@ CREATE TABLE read_positions_cache (
     UNIQUE(user_did, rkey)
 );
 
-CREATE INDEX idx_read_positions_cache_user ON read_positions_cache(user_did);
-CREATE INDEX idx_read_positions_cache_guid ON read_positions_cache(user_did, item_guid);
-CREATE INDEX idx_read_positions_cache_unsynced ON read_positions_cache(synced_at) WHERE synced_at IS NULL;
+CREATE INDEX IF NOT EXISTS idx_read_positions_cache_user ON read_positions_cache(user_did);
+CREATE INDEX IF NOT EXISTS idx_read_positions_cache_guid ON read_positions_cache(user_did, item_guid);
+CREATE INDEX IF NOT EXISTS idx_read_positions_cache_unsynced ON read_positions_cache(synced_at) WHERE synced_at IS NULL;
diff --git a/backend/migrations/0005_share_guid_published.sql b/backend/migrations/0005_share_guid_published.sql
@@ -4,4 +4,4 @@ ALTER TABLE shares ADD COLUMN item_published_at INTEGER;
 ALTER TABLE shares ADD COLUMN feed_url TEXT;
 
 -- Index for looking up shares by guid
-CREATE INDEX idx_shares_item_guid ON shares(item_guid);
+CREATE INDEX IF NOT EXISTS idx_shares_item_guid ON shares(item_guid);
diff --git a/backend/migrations/0011_feed_items.sql b/backend/migrations/0011_feed_items.sql
@@ -1,5 +1,5 @@
 -- Store individual feed items for efficient querying
-CREATE TABLE feed_items (
+CREATE TABLE IF NOT EXISTS feed_items (
     id INTEGER PRIMARY KEY AUTOINCREMENT,
     feed_url TEXT NOT NULL,
     guid TEXT NOT NULL,
@@ -16,13 +16,13 @@ CREATE TABLE feed_items (
 );
 
 -- Index for fetching items by feed
-CREATE INDEX idx_feed_items_feed_url ON feed_items(feed_url);
+CREATE INDEX IF NOT EXISTS idx_feed_items_feed_url ON feed_items(feed_url);
 
 -- Index for cross-feed queries sorted by date
-CREATE INDEX idx_feed_items_published_at ON feed_items(published_at DESC);
+CREATE INDEX IF NOT EXISTS idx_feed_items_published_at ON feed_items(published_at DESC);
 
 -- Composite index for feed-specific date queries
-CREATE INDEX idx_feed_items_feed_published ON feed_items(feed_url, published_at DESC);
+CREATE INDEX IF NOT EXISTS idx_feed_items_feed_published ON feed_items(feed_url, published_at DESC);
 
 -- Index for looking up items by URL (for social features)
-CREATE INDEX idx_feed_items_url ON feed_items(url);
+CREATE INDEX IF NOT EXISTS idx_feed_items_url ON feed_items(url);
diff --git a/backend/migrations/0012_inapp_follows.sql b/backend/migrations/0012_inapp_follows.sql
@@ -1,5 +1,5 @@
 -- In-app follows table (separate from Bluesky follows in follows_cache)
-CREATE TABLE inapp_follows (
+CREATE TABLE IF NOT EXISTS inapp_follows (
     id INTEGER PRIMARY KEY AUTOINCREMENT,
     follower_did TEXT NOT NULL,
     following_did TEXT NOT NULL,
@@ -9,6 +9,6 @@ CREATE TABLE inapp_follows (
     UNIQUE(follower_did, following_did)
 );
 
-CREATE INDEX idx_inapp_follows_follower ON inapp_follows(follower_did);
-CREATE INDEX idx_inapp_follows_following ON inapp_follows(following_did);
-CREATE INDEX idx_inapp_follows_rkey ON inapp_follows(follower_did, rkey);
+CREATE INDEX IF NOT EXISTS idx_inapp_follows_follower ON inapp_follows(follower_did);
+CREATE INDEX IF NOT EXISTS idx_inapp_follows_following ON inapp_follows(following_did);
+CREATE INDEX IF NOT EXISTS idx_inapp_follows_rkey ON inapp_follows(follower_did, rkey);
diff --git a/backend/migrations/0013_share_read_positions_cache.sql b/backend/migrations/0013_share_read_positions_cache.sql
@@ -1,6 +1,6 @@
 -- Cache of share read positions (synced from AT Protocol)
 -- Used to dedupe before syncing to PDS
-CREATE TABLE share_read_positions_cache (
+CREATE TABLE IF NOT EXISTS share_read_positions_cache (
     id INTEGER PRIMARY KEY AUTOINCREMENT,
     user_did TEXT NOT NULL,
     rkey TEXT NOT NULL,
@@ -17,5 +17,5 @@ CREATE TABLE share_read_positions_cache (
     UNIQUE(user_did, share_uri)
 );
 
-CREATE INDEX idx_share_read_positions_cache_user ON share_read_positions_cache(user_did);
-CREATE INDEX idx_share_read_positions_cache_share ON share_read_positions_cache(user_did, share_uri);
+CREATE INDEX IF NOT EXISTS idx_share_read_positions_cache_user ON share_read_positions_cache(user_did);
+CREATE INDEX IF NOT EXISTS idx_share_read_positions_cache_share ON share_read_positions_cache(user_did, share_uri);
diff --git a/backend/migrations/0016_rate_limits.sql b/backend/migrations/0016_rate_limits.sql
@@ -0,0 +1,11 @@
+-- Rate limiting table for per-user API request tracking
+CREATE TABLE IF NOT EXISTS rate_limits (
+  did TEXT NOT NULL,
+  endpoint TEXT NOT NULL,
+  window_start INTEGER NOT NULL,
+  count INTEGER DEFAULT 1,
+  PRIMARY KEY (did, endpoint, window_start)
+);
+
+-- Index for efficient cleanup of old rate limit records
+CREATE INDEX IF NOT EXISTS idx_rate_limits_cleanup ON rate_limits(window_start);
diff --git a/backend/src/index.ts b/backend/src/index.ts
@@ -1,6 +1,6 @@
 import type { Env } from './types';
 import { handleAuthLogin, handleAuthCallback, handleAuthLogout, handleAuthMe, handleClientMetadata } from './routes/auth';
-import { handleFeedFetch, handleCachedFeedFetch, handleFeedDiscover, handleArticleFetch, handleFeedStatusBatch } from './routes/feeds';
+import { handleFeedFetch, handleCachedFeedFetch, handleFeedDiscover, handleArticleFetch, handleFeedStatusBatch, handleBatchFeedFetch } from './routes/feeds';
 import { handleItemsList, handleItemsRecent, handleItemGet, handleItemsByFeed } from './routes/items';
 import { handleSocialFeed, handleSyncFollows, handleFollowedUsers, handlePopularShares } from './routes/social';
 import { handleGetMyShares } from './routes/shares';
@@ -13,6 +13,7 @@ import { pollJetstream } from './services/jetstream-poller';
 import { pollJetstreamFollows } from './services/jetstream-follows-poller';
 import { pollJetstreamInappFollows } from './services/jetstream-inapp-follows-poller';
 import { syncReadPositionsToPds } from './services/read-positions-pds-sync';
+import { checkRateLimit, cleanupRateLimits, getRateLimitConfig } from './services/rate-limit';
 
 export { RealtimeHub } from './durable-objects/realtime-hub';
 
@@ -40,6 +41,22 @@ export default {
     const session = await getSessionFromRequest(request, env);
     if (session) {
       ctx.waitUntil(updateUserActivity(env, session.did));
+
+      // Check rate limit for authenticated requests
+      const rateLimit = await checkRateLimit(env, session.did, url.pathname);
+      if (!rateLimit.allowed) {
+        const config = getRateLimitConfig(url.pathname);
+        return new Response(JSON.stringify({ error: 'Rate limit exceeded' }), {
+          status: 429,
+          headers: {
+            ...headers,
+            'Content-Type': 'application/json',
+            'Retry-After': String(rateLimit.retryAfter || 60),
+            'X-RateLimit-Limit': String(config.limit),
+            'X-RateLimit-Remaining': '0',
+          },
+        });
+      }
     }
 
     try {
@@ -82,6 +99,9 @@ export default {
         case url.pathname === '/api/feeds/status':
           response = await handleFeedStatusBatch(request, env);
           break;
+        case url.pathname === '/api/feeds/batch':
+          response = await handleBatchFeedFetch(request, env);
+          break;
 
         // Item routes (individual feed items)
         case url.pathname === '/api/items':
@@ -273,6 +293,16 @@ export default {
       }
     }
 
+    // Clean up rate limit records (every minute)
+    try {
+      const rateLimitDeleted = await cleanupRateLimits(env);
+      if (rateLimitDeleted > 0) {
+        console.log(`[Cron] Rate limit cleanup: deleted ${rateLimitDeleted} records`);
+      }
+    } catch (error) {
+      console.error('[Cron] Rate limit cleanup failed:', error);
+    }
+
     // Clean up expired D1 data (once per hour)
     if (minute === 0) {
       try {
diff --git a/backend/src/routes/feeds.ts b/backend/src/routes/feeds.ts
diff --git a/backend/src/services/rate-limit.ts b/backend/src/services/rate-limit.ts
diff --git a/frontend/src/lib/services/api.ts b/frontend/src/lib/services/api.ts
diff --git a/frontend/src/lib/stores/subscriptions.svelte.ts b/frontend/src/lib/stores/subscriptions.svelte.ts
