feat: Add session management and rate limiting

- Stores and restores session for persistent login - Implements dynamic
rate limiting from ATP headers - Adds Prometheus metrics for rate limit
state and waits - Refactors authentication flow for session persistence

compose.yaml

···
     # after a restart, preventing it from reprocessing old events or skipping new ones.
     volumes:
       - ./cursor.txt:/app/cursor.txt
+      - ./.session:/app/.session
 
     environment:
       - NODE_ENV=production

src/agent.ts

···
 import { Agent, setGlobalDispatcher } from "undici";
 import { AtpAgent } from "@atproto/api";
 import { BSKY_HANDLE, BSKY_PASSWORD, OZONE_PDS } from "./config.js";
+import { loadSession, saveSession, type SessionData } from "./session.js";
+import { updateRateLimitState } from "./limits.js";
+import { logger } from "./logger.js";
 
 setGlobalDispatcher(new Agent({ connect: { timeout: 20_000 } }));
 
+const customFetch: typeof fetch = async (input, init) => {
+  const response = await fetch(input, init);
+
+  // Extract rate limit headers from ATP responses
+  const limitHeader = response.headers.get("ratelimit-limit");
+  const remainingHeader = response.headers.get("ratelimit-remaining");
+  const resetHeader = response.headers.get("ratelimit-reset");
+  const policyHeader = response.headers.get("ratelimit-policy");
+
+  if (limitHeader && remainingHeader && resetHeader) {
+    updateRateLimitState({
+      limit: parseInt(limitHeader, 10),
+      remaining: parseInt(remainingHeader, 10),
+      reset: parseInt(resetHeader, 10),
+      policy: policyHeader || undefined,
+    });
+  }
+
+  return response;
+};
+
 export const agent = new AtpAgent({
   service: `https://${OZONE_PDS}`,
+  fetch: customFetch,
 });
-export const login = () =>
-  agent.login({
-    identifier: BSKY_HANDLE,
-    password: BSKY_PASSWORD,
-  });
+
+const JWT_LIFETIME_MS = 2 * 60 * 60 * 1000; // 2 hours (typical ATP JWT lifetime)
+const REFRESH_AT_PERCENT = 0.8; // Refresh at 80% of lifetime
+let refreshTimer: NodeJS.Timeout | null = null;
 
-export const isLoggedIn = login()
-  .then(() => true)
+async function refreshSession(): Promise<void> {
+  try {
+    logger.info("Refreshing session tokens");
+    await agent.resumeSession(agent.session!);
+
+    if (agent.session) {
+      saveSession(agent.session as SessionData);
+      scheduleSessionRefresh();
+    }
+  } catch (error) {
+    logger.error({ error }, "Failed to refresh session, will re-authenticate");
+    await performLogin();
+  }
+}
+
+function scheduleSessionRefresh(): void {
+  if (refreshTimer) {
+    clearTimeout(refreshTimer);
+  }
+
+  const refreshIn = JWT_LIFETIME_MS * REFRESH_AT_PERCENT;
+  logger.debug(`Scheduling session refresh in ${(refreshIn / 1000 / 60).toFixed(1)} minutes`);
+
+  refreshTimer = setTimeout(() => {
+    refreshSession().catch((error) => {
+      logger.error({ error }, "Scheduled session refresh failed");
+    });
+  }, refreshIn);
+}
+
+async function performLogin(): Promise<boolean> {
+  try {
+    logger.info("Performing fresh login");
+    const response = await agent.login({
+      identifier: BSKY_HANDLE,
+      password: BSKY_PASSWORD,
+    });
+
+    if (response.success && agent.session) {
+      saveSession(agent.session as SessionData);
+      scheduleSessionRefresh();
+      logger.info("Login successful, session saved");
+      return true;
+    }
+
+    logger.error("Login failed: no session returned");
+    return false;
+  } catch (error) {
+    logger.error({ error }, "Login failed");
+    return false;
+  }
+}
+
+async function authenticate(): Promise<boolean> {
+  const savedSession = loadSession();
+
+  if (savedSession) {
+    try {
+      logger.info("Attempting to resume saved session");
+      await agent.resumeSession(savedSession);
+
+      // Verify session is still valid with a lightweight call
+      await agent.getProfile({ actor: savedSession.did });
+
+      logger.info("Session resumed successfully");
+      scheduleSessionRefresh();
+      return true;
+    } catch (error) {
+      logger.warn({ error }, "Saved session invalid, will re-authenticate");
+    }
+  }
+
+  return performLogin();
+}
+
+export const login = authenticate;
+export const isLoggedIn = authenticate()
+  .then((success) => success)
   .catch(() => false);

src/limits.ts

···
 import { pRateLimit } from "p-ratelimit";
+import { logger } from "./logger.js";
+import { Counter, Gauge, Histogram } from "prom-client";
 
-// TypeScript
+interface RateLimitState {
+  limit: number;
+  remaining: number;
+  reset: number; // Unix timestamp in seconds
+  policy?: string;
+}
+
+// Conservative defaults based on previous static configuration
+// Will be replaced with dynamic values from ATP response headers
+let rateLimitState: RateLimitState = {
+  limit: 280,
+  remaining: 280,
+  reset: Math.floor(Date.now() / 1000) + 30,
+};
 
-// create a rate limiter that allows up to 30 API calls per second,
-// with max concurrency of 10
+const SAFETY_BUFFER = 5; // Keep this many requests in reserve (reduced from 20)
+const CONCURRENCY = 24; // Reduced from 48 to prevent rapid depletion
 
-export const limit = pRateLimit({
-  interval: 30000, // 1000 ms == 1 second
-  rate: 280, // 30 API calls per interval
-  concurrency: 48, // no more than 10 running at once
-  maxDelay: 0, // an API call delayed > 30 sec is rejected
+// Metrics
+const rateLimitWaitsTotal = new Counter({
+  name: "rate_limit_waits_total",
+  help: "Total number of times rate limit wait was triggered",
 });
+
+const rateLimitWaitDuration = new Histogram({
+  name: "rate_limit_wait_duration_seconds",
+  help: "Duration of rate limit waits in seconds",
+  buckets: [0.1, 0.5, 1, 5, 10, 30, 60],
+});
+
+const rateLimitRemaining = new Gauge({
+  name: "rate_limit_remaining",
+  help: "Current remaining rate limit",
+});
+
+const rateLimitTotal = new Gauge({
+  name: "rate_limit_total",
+  help: "Total rate limit from headers",
+});
+
+const concurrentRequestsGauge = new Gauge({
+  name: "concurrent_requests",
+  help: "Current number of concurrent requests",
+});
+
+// Use p-ratelimit purely for concurrency management
+const concurrencyLimiter = pRateLimit({
+  interval: 1000,
+  rate: 10000, // Very high rate, we manage rate limiting separately
+  concurrency: CONCURRENCY,
+  maxDelay: 0,
+});
+
+export function getRateLimitState(): RateLimitState {
+  return { ...rateLimitState };
+}
+
+export function updateRateLimitState(state: Partial<RateLimitState>): void {
+  rateLimitState = { ...rateLimitState, ...state };
+
+  // Update Prometheus metrics
+  if (state.remaining !== undefined) {
+    rateLimitRemaining.set(state.remaining);
+  }
+  if (state.limit !== undefined) {
+    rateLimitTotal.set(state.limit);
+  }
+
+  logger.debug(
+    {
+      limit: rateLimitState.limit,
+      remaining: rateLimitState.remaining,
+      resetIn: rateLimitState.reset - Math.floor(Date.now() / 1000),
+    },
+    "Rate limit state updated"
+  );
+}
+
+async function awaitRateLimit(): Promise<void> {
+  const state = getRateLimitState();
+  const now = Math.floor(Date.now() / 1000);
+
+  // Only wait if we're critically low
+  if (state.remaining <= SAFETY_BUFFER) {
+    rateLimitWaitsTotal.inc();
+
+    const delaySeconds = Math.max(0, state.reset - now);
+    const delayMs = delaySeconds * 1000;
+
+    if (delayMs > 0) {
+      logger.warn(
+        `Rate limit critical (${state.remaining}/${state.limit} remaining). Waiting ${delaySeconds}s until reset...`
+      );
+
+      const waitStart = Date.now();
+      await new Promise((resolve) => setTimeout(resolve, delayMs));
+      const waitDuration = (Date.now() - waitStart) / 1000;
+      rateLimitWaitDuration.observe(waitDuration);
+
+      // Don't manually reset state - let the next API response update it
+      logger.info("Rate limit wait complete, resuming requests");
+    }
+  }
+}
+
+export async function limit<T>(fn: () => Promise<T>): Promise<T> {
+  return concurrencyLimiter(async () => {
+    concurrentRequestsGauge.inc();
+    try {
+      await awaitRateLimit();
+      return await fn();
+    } finally {
+      concurrentRequestsGauge.dec();
+    }
+  });
+}

src/session.ts

···
+import { readFileSync, writeFileSync, unlinkSync, chmodSync, existsSync } from "node:fs";
+import { join } from "node:path";
+import { logger } from "./logger.js";
+
+const SESSION_FILE_PATH = join(process.cwd(), ".session");
+
+export interface SessionData {
+  accessJwt: string;
+  refreshJwt: string;
+  did: string;
+  handle: string;
+  email?: string;
+  emailConfirmed?: boolean;
+  emailAuthFactor?: boolean;
+  active: boolean;
+  status?: string;
+}
+
+export function loadSession(): SessionData | null {
+  try {
+    if (!existsSync(SESSION_FILE_PATH)) {
+      logger.debug("No session file found");
+      return null;
+    }
+
+    const data = readFileSync(SESSION_FILE_PATH, "utf-8");
+    const session = JSON.parse(data) as SessionData;
+
+    if (!session.accessJwt || !session.refreshJwt || !session.did) {
+      logger.warn("Session file is missing required fields, ignoring");
+      return null;
+    }
+
+    logger.info("Loaded existing session from file");
+    return session;
+  } catch (error) {
+    logger.error({ error }, "Failed to load session file, will authenticate fresh");
+    return null;
+  }
+}
+
+export function saveSession(session: SessionData): void {
+  try {
+    const data = JSON.stringify(session, null, 2);
+    writeFileSync(SESSION_FILE_PATH, data, "utf-8");
+    chmodSync(SESSION_FILE_PATH, 0o600);
+    logger.info("Session saved to file");
+  } catch (error) {
+    logger.error({ error }, "Failed to save session to file");
+  }
+}
+
+export function clearSession(): void {
+  try {
+    if (existsSync(SESSION_FILE_PATH)) {
+      unlinkSync(SESSION_FILE_PATH);
+      logger.info("Session file cleared");
+    }
+  } catch (error) {
+    logger.error({ error }, "Failed to clear session file");
+  }
+}