Merge pull request #27 from skywatch-bsky/feat/env-validation

Feat/env validation

.claude/settings.local.json

···
       "Bash(bun run lint:*)",
       "mcp__git-mcp-server__git_commit",
       "mcp__git-mcp-server__git_push",
-      "Bash(bunx eslint:*)",
+      "Bash(bunx tsc:*)",
+      "Bash(DID=did:test OZONE_URL=http://test OZONE_PDS=http://test BSKY_HANDLE=test.bsky.social BSKY_PASSWORD=testpass bun --eval \"import(''./src/validateEnv.js'').then(m => m.validateEnvironment())\")",
       "mcp__git-mcp-server__git_add"
     ],
     "ask": [

eslint.config.mjs

···
 
       // Style preferences
       "@stylistic/indent": ["error", 2],
-      "@stylistic/quotes": ["error", "single"],
+      "@stylistic/quotes": ["error", "double"],
       "@stylistic/semi": ["error", "always"],
       //"@stylistic/comma-dangle": ["error", "es5"],
       "@stylistic/object-curly-spacing": ["error", "always"],

src/main.ts

···
-import fs from 'node:fs';
+import fs from "node:fs";
 
 import type {
   CommitCreateEvent,
   CommitUpdateEvent,
-  IdentityEvent } from '@skyware/jetstream';
-import {
-  Jetstream,
-} from '@skyware/jetstream';
-
+  IdentityEvent,
+} from "@skyware/jetstream";
+import { Jetstream } from "@skyware/jetstream";
 
-import { checkHandle } from './checkHandles.js';
-import { checkPosts } from './checkPosts.js';
-import { checkDescription, checkDisplayName } from './checkProfiles.js';
-import { checkStarterPack, checkNewStarterPack } from './checkStarterPack.js';
+import { checkHandle } from "./checkHandles.js";
+import { checkPosts } from "./checkPosts.js";
+import { checkDescription, checkDisplayName } from "./checkProfiles.js";
+import { checkStarterPack, checkNewStarterPack } from "./checkStarterPack.js";
 import {
   CURSOR_UPDATE_INTERVAL,
   FIREHOSE_URL,
   METRICS_PORT,
   WANTED_COLLECTION,
-} from './config.js';
-import logger from './logger.js';
-import { startMetricsServer } from './metrics.js';
-import type { Post, LinkFeature } from './types.js';
+} from "./config.js";
+import { validateEnvironment } from "./validateEnv.js";
+import logger from "./logger.js";
+import { startMetricsServer } from "./metrics.js";
+import type { Post, LinkFeature } from "./types.js";
+
+validateEnvironment();
 
 let cursor = 0;
 let cursorUpdateInterval: NodeJS.Timeout;
···
 }
 
 try {
-  logger.info('Trying to read cursor from cursor.txt...');
-  cursor = Number(fs.readFileSync('cursor.txt', 'utf8'));
-  logger.info(`Cursor found: ${cursor.toString()} (${epochUsToDateTime(cursor)})`);
+  logger.info("Trying to read cursor from cursor.txt...");
+  cursor = Number(fs.readFileSync("cursor.txt", "utf8"));
+  logger.info(
+    `Cursor found: ${cursor.toString()} (${epochUsToDateTime(cursor)})`,
+  );
 } catch (error) {
-  if (error instanceof Error && 'code' in error && error.code === 'ENOENT') {
+  if (error instanceof Error && "code" in error && error.code === "ENOENT") {
     cursor = Math.floor(Date.now() * 1000);
     logger.info(
       `Cursor not found in cursor.txt, setting cursor to: ${cursor.toString()} (${epochUsToDateTime(cursor)})`,
     );
-    fs.writeFileSync('cursor.txt', cursor.toString(), 'utf8');
+    fs.writeFileSync("cursor.txt", cursor.toString(), "utf8");
   } else {
     logger.error(error);
     process.exit(1);
···
   cursor,
 });
 
-jetstream.on('open', () => {
+jetstream.on("open", () => {
   if (jetstream.cursor) {
     logger.info(
       `Connected to Jetstream at ${FIREHOSE_URL} with cursor ${jetstream.cursor.toString()} (${epochUsToDateTime(jetstream.cursor)})`,
···
       logger.info(
         `Cursor updated to: ${jetstream.cursor.toString()} (${epochUsToDateTime(jetstream.cursor)})`,
       );
-      fs.writeFile('cursor.txt', jetstream.cursor.toString(), (err) => {
+      fs.writeFile("cursor.txt", jetstream.cursor.toString(), (err) => {
         if (err) logger.error(err);
       });
     }
   }, CURSOR_UPDATE_INTERVAL);
 });
 
-jetstream.on('close', () => {
+jetstream.on("close", () => {
   clearInterval(cursorUpdateInterval);
-  logger.info('Jetstream connection closed.');
+  logger.info("Jetstream connection closed.");
 });
 
-jetstream.on('error', (error) => {
+jetstream.on("error", (error) => {
   logger.error(`Jetstream error: ${error.message}`);
 });
 
 // Check for post updates
 
 jetstream.onCreate(
-  'app.bsky.feed.post',
-  (event: CommitCreateEvent<'app.bsky.feed.post'>) => {
+  "app.bsky.feed.post",
+  (event: CommitCreateEvent<"app.bsky.feed.post">) => {
     try {
       const atURI = `at://${event.did}/app.bsky.feed.post/${event.commit.rkey}`;
-      const hasFacets = Object.hasOwn(event.commit.record, 'facets');
-      const hasText = Object.hasOwn(event.commit.record, 'text');
+      const hasFacets = Object.hasOwn(event.commit.record, "facets");
+      const hasText = Object.hasOwn(event.commit.record, "text");
 
       const tasks: Promise<void>[] = [];
 
···
       if (hasFacets && event.commit.record.facets) {
         const hasLinkType = event.commit.record.facets.some((facet) =>
           facet.features.some(
-            (feature) => feature.$type === 'app.bsky.richtext.facet#link',
+            (feature) => feature.$type === "app.bsky.richtext.facet#link",
           ),
         );
 
         if (hasLinkType) {
-          const urls = event.commit.record.facets.flatMap((facet) =>
-            facet.features.filter(
-              (feature) => feature.$type === 'app.bsky.richtext.facet#link',
-            ),
-          )
+          const urls = event.commit.record.facets
+            .flatMap((facet) =>
+              facet.features.filter(
+                (feature) => feature.$type === "app.bsky.richtext.facet#link",
+              ),
+            )
             .map((feature: LinkFeature) => feature.uri);
 
           urls.forEach((url) => {
···
                 cid: event.commit.cid,
               },
             ];
-            tasks.push(checkPosts(posts).catch((error: unknown) => {
-              logger.error(`Error checking post links for ${event.did}:`, error);
-            }));
+            tasks.push(
+              checkPosts(posts).catch((error: unknown) => {
+                logger.error(
+                  `Error checking post links for ${event.did}:`,
+                  error,
+                );
+              }),
+            );
           });
         }
       } else if (hasText && event.commit.record.text) {
···
             cid: event.commit.cid,
           },
         ];
-        tasks.push(checkPosts(posts).catch((error: unknown) => {
-          logger.error(`Error checking post text for ${event.did}:`, error);
-        }));
+        tasks.push(
+          checkPosts(posts).catch((error: unknown) => {
+            logger.error(`Error checking post text for ${event.did}:`, error);
+          }),
+        );
       }
 
       // Wait for all tasks to complete
···
 
 // Check for profile updates
 jetstream.onUpdate(
-  'app.bsky.actor.profile',
-  (event: CommitUpdateEvent<'app.bsky.actor.profile'>) => {
+  "app.bsky.actor.profile",
+  (event: CommitUpdateEvent<"app.bsky.actor.profile">) => {
     try {
       const tasks: Promise<void>[] = [];
 
       if (event.commit.record.displayName || event.commit.record.description) {
-        const displayName = event.commit.record.displayName ?? '';
-        const description = event.commit.record.description ?? '';
-        
+        const displayName = event.commit.record.displayName ?? "";
+        const description = event.commit.record.description ?? "";
+
         tasks.push(
-          checkDescription(event.did, event.time_us, displayName, description)
-            .catch((error: unknown) => {
-              logger.error(`Error checking profile description for ${event.did}:`, error);
-            })
+          checkDescription(
+            event.did,
+            event.time_us,
+            displayName,
+            description,
+          ).catch((error: unknown) => {
+            logger.error(
+              `Error checking profile description for ${event.did}:`,
+              error,
+            );
+          }),
         );
-        
+
         tasks.push(
-          checkDisplayName(event.did, event.time_us, displayName, description)
-            .catch((error: unknown) => {
-              logger.error(`Error checking profile display name for ${event.did}:`, error);
-            })
+          checkDisplayName(
+            event.did,
+            event.time_us,
+            displayName,
+            description,
+          ).catch((error: unknown) => {
+            logger.error(
+              `Error checking profile display name for ${event.did}:`,
+              error,
+            );
+          }),
         );
       }
 
       if (event.commit.record.joinedViaStarterPack) {
         tasks.push(
-          checkStarterPack(event.did, event.time_us, event.commit.record.joinedViaStarterPack.uri)
-            .catch((error: unknown) => {
-              logger.error(`Error checking starter pack for ${event.did}:`, error);
-            })
+          checkStarterPack(
+            event.did,
+            event.time_us,
+            event.commit.record.joinedViaStarterPack.uri,
+          ).catch((error: unknown) => {
+            logger.error(
+              `Error checking starter pack for ${event.did}:`,
+              error,
+            );
+          }),
         );
       }
 
···
         void Promise.allSettled(tasks);
       }
     } catch (error: unknown) {
-      logger.error(`Error processing profile update event for ${event.did}:`, error);
+      logger.error(
+        `Error processing profile update event for ${event.did}:`,
+        error,
+      );
     }
   },
 );
···
 // Check for profile updates
 
 jetstream.onCreate(
-  'app.bsky.actor.profile',
-  (event: CommitCreateEvent<'app.bsky.actor.profile'>) => {
+  "app.bsky.actor.profile",
+  (event: CommitCreateEvent<"app.bsky.actor.profile">) => {
     try {
       const tasks: Promise<void>[] = [];
 
       if (event.commit.record.displayName || event.commit.record.description) {
-        const displayName = event.commit.record.displayName ?? '';
-        const description = event.commit.record.description ?? '';
-        
+        const displayName = event.commit.record.displayName ?? "";
+        const description = event.commit.record.description ?? "";
+
         tasks.push(
-          checkDescription(event.did, event.time_us, displayName, description)
-            .catch((error: unknown) => {
-              logger.error(`Error checking profile description for ${event.did}:`, error);
-            })
+          checkDescription(
+            event.did,
+            event.time_us,
+            displayName,
+            description,
+          ).catch((error: unknown) => {
+            logger.error(
+              `Error checking profile description for ${event.did}:`,
+              error,
+            );
+          }),
         );
-        
+
         tasks.push(
-          checkDisplayName(event.did, event.time_us, displayName, description)
-            .catch((error: unknown) => {
-              logger.error(`Error checking profile display name for ${event.did}:`, error);
-            })
+          checkDisplayName(
+            event.did,
+            event.time_us,
+            displayName,
+            description,
+          ).catch((error: unknown) => {
+            logger.error(
+              `Error checking profile display name for ${event.did}:`,
+              error,
+            );
+          }),
         );
 
         if (event.commit.record.joinedViaStarterPack) {
           tasks.push(
-            checkStarterPack(event.did, event.time_us, event.commit.record.joinedViaStarterPack.uri)
-              .catch((error: unknown) => {
-                logger.error(`Error checking starter pack for ${event.did}:`, error);
-              })
+            checkStarterPack(
+              event.did,
+              event.time_us,
+              event.commit.record.joinedViaStarterPack.uri,
+            ).catch((error: unknown) => {
+              logger.error(
+                `Error checking starter pack for ${event.did}:`,
+                error,
+              );
+            }),
           );
         }
 
···
         }
       }
     } catch (error: unknown) {
-      logger.error(`Error processing profile creation event for ${event.did}:`, error);
+      logger.error(
+        `Error processing profile creation event for ${event.did}:`,
+        error,
+      );
     }
   },
 );
 
 jetstream.onCreate(
-  'app.bsky.graph.starterpack',
-  (event: CommitCreateEvent<'app.bsky.graph.starterpack'>) => {
+  "app.bsky.graph.starterpack",
+  (event: CommitCreateEvent<"app.bsky.graph.starterpack">) => {
     try {
       const atURI = `at://${event.did}/app.bsky.feed.post/${event.commit.rkey}`;
       const { name, description } = event.commit.record;
···
         name,
         description,
       ).catch((error: unknown) => {
-        logger.error(`Error checking new starter pack for ${event.did}:`, error);
+        logger.error(
+          `Error checking new starter pack for ${event.did}:`,
+          error,
+        );
       });
     } catch (error: unknown) {
-      logger.error(`Error processing starter pack creation event for ${event.did}:`, error);
+      logger.error(
+        `Error processing starter pack creation event for ${event.did}:`,
+        error,
+      );
     }
   },
 );
 
 jetstream.onUpdate(
-  'app.bsky.graph.starterpack',
-  (event: CommitUpdateEvent<'app.bsky.graph.starterpack'>) => {
+  "app.bsky.graph.starterpack",
+  (event: CommitUpdateEvent<"app.bsky.graph.starterpack">) => {
     try {
       const atURI = `at://${event.did}/app.bsky.feed.post/${event.commit.rkey}`;
       const { name, description } = event.commit.record;
···
         name,
         description,
       ).catch((error: unknown) => {
-        logger.error(`Error checking updated starter pack for ${event.did}:`, error);
+        logger.error(
+          `Error checking updated starter pack for ${event.did}:`,
+          error,
+        );
       });
     } catch (error: unknown) {
-      logger.error(`Error processing starter pack update event for ${event.did}:`, error);
+      logger.error(
+        `Error processing starter pack update event for ${event.did}:`,
+        error,
+      );
     }
   },
 );
 
 // Check for handle updates
-jetstream.on('identity', (event: IdentityEvent) => {
+jetstream.on("identity", (event: IdentityEvent) => {
   try {
     if (event.identity.handle) {
-      void checkHandle(event.identity.did, event.identity.handle, event.time_us)
-        .catch((error: unknown) => {
-          logger.error(`Error checking handle for ${event.identity.did}:`, error);
-        });
+      void checkHandle(
+        event.identity.did,
+        event.identity.handle,
+        event.time_us,
+      ).catch((error: unknown) => {
+        logger.error(`Error checking handle for ${event.identity.did}:`, error);
+      });
     }
   } catch (error: unknown) {
-    logger.error(`Error processing identity event for ${event.identity.did}:`, error);
+    logger.error(
+      `Error processing identity event for ${event.identity.did}:`,
+      error,
+    );
   }
 });
 
···
   metricsServer = startMetricsServer(METRICS_PORT);
   logger.info(`Metrics server started on port ${METRICS_PORT.toString()}`);
 } catch (error: unknown) {
-  logger.error('Failed to start metrics server:', error);
+  logger.error("Failed to start metrics server:", error);
   process.exit(1);
 }
 
···
 // Start jetstream with error handling
 try {
   jetstream.start();
-  logger.info('Jetstream started successfully');
+  logger.info("Jetstream started successfully");
 } catch (error: unknown) {
-  logger.error('Failed to start jetstream:', error);
+  logger.error("Failed to start jetstream:", error);
   process.exit(1);
 }
 
 function shutdown() {
   try {
-    logger.info('Shutting down gracefully...');
+    logger.info("Shutting down gracefully...");
     if (jetstream.cursor) {
-      fs.writeFileSync('cursor.txt', jetstream.cursor.toString(), 'utf8');
+      fs.writeFileSync("cursor.txt", jetstream.cursor.toString(), "utf8");
     }
     jetstream.close();
     if (metricsServer) {
       metricsServer.close(() => {
-        logger.info('Metrics server closed');
+        logger.info("Metrics server closed");
       });
     }
-    logger.info('Shutdown completed successfully');
+    logger.info("Shutdown completed successfully");
   } catch (error: unknown) {
-    logger.error('Error shutting down gracefully:', error);
+    logger.error("Error shutting down gracefully:", error);
     process.exit(1);
   }
 }
 
 // Global error handlers
-process.on('unhandledRejection', (reason, promise) => {
-  logger.error('Unhandled Promise Rejection at:', promise, 'reason:', reason);
+process.on("unhandledRejection", (reason, promise) => {
+  logger.error("Unhandled Promise Rejection at:", promise, "reason:", reason);
   // Don't exit the process for unhandled rejections, just log them
 });
 
-process.on('uncaughtException', (error) => {
-  logger.error('Uncaught Exception:', error);
+process.on("uncaughtException", (error) => {
+  logger.error("Uncaught Exception:", error);
   shutdown();
 });
 
-process.on('SIGINT', shutdown);
-process.on('SIGTERM', shutdown);
+process.on("SIGINT", shutdown);
+process.on("SIGTERM", shutdown);

src/validateEnv.ts

···
+import logger from './logger.js';
+
+interface EnvironmentVariable {
+  name: string;
+  required: boolean;
+  description: string;
+  validator?: (value: string) => boolean;
+}
+
+const ENV_VARIABLES: EnvironmentVariable[] = [
+  {
+    name: 'DID',
+    required: true,
+    description: 'Moderator DID for labeling operations',
+    validator: (value) => value.startsWith('did:'),
+  },
+  {
+    name: 'OZONE_URL',
+    required: true,
+    description: 'Ozone server URL',
+    validator: (value) => value.includes('.') && value.length > 3,
+  },
+  {
+    name: 'OZONE_PDS',
+    required: true,
+    description: 'Ozone PDS URL',
+    validator: (value) => value.includes('.') && value.length > 3,
+  },
+  {
+    name: 'BSKY_HANDLE',
+    required: true,
+    description: 'Bluesky handle for authentication',
+    validator: (value) => value.includes('.'),
+  },
+  {
+    name: 'BSKY_PASSWORD',
+    required: true,
+    description: 'Bluesky password for authentication',
+    validator: (value) => value.length > 0,
+  },
+  {
+    name: 'HOST',
+    required: false,
+    description: 'Host address for the server (defaults to 127.0.0.1)',
+  },
+  {
+    name: 'PORT',
+    required: false,
+    description: 'Port for the main server (defaults to 4100)',
+    validator: (value) => !isNaN(Number(value)) && Number(value) > 0,
+  },
+  {
+    name: 'METRICS_PORT',
+    required: false,
+    description: 'Port for metrics server (defaults to 4101)',
+    validator: (value) => !isNaN(Number(value)) && Number(value) > 0,
+  },
+  {
+    name: 'FIREHOSE_URL',
+    required: false,
+    description: 'Jetstream firehose WebSocket URL',
+    validator: (value) => value.startsWith('ws'),
+  },
+  {
+    name: 'CURSOR_UPDATE_INTERVAL',
+    required: false,
+    description: 'Cursor update interval in milliseconds (defaults to 60000)',
+    validator: (value) => !isNaN(Number(value)) && Number(value) > 0,
+  },
+  {
+    name: 'LABEL_LIMIT',
+    required: false,
+    description: 'Rate limit for labeling operations',
+    validator: (value) => {
+      // Allow "number * number" format or plain numbers
+      const multiplyMatch = /^(\d+)\s*\*\s*(\d+)$/.exec(value);
+      if (multiplyMatch) {
+        const result = Number(multiplyMatch[1]) * Number(multiplyMatch[2]);
+        return result > 0;
+      }
+      return !isNaN(Number(value)) && Number(value) > 0;
+    },
+  },
+  {
+    name: 'LABEL_LIMIT_WAIT',
+    required: false,
+    description: 'Wait time between rate limited operations',
+    validator: (value) => {
+      // Allow "number * number" format or plain numbers
+      const multiplyMatch = /^(\d+)\s*\*\s*(\d+)$/.exec(value);
+      if (multiplyMatch) {
+        const result = Number(multiplyMatch[1]) * Number(multiplyMatch[2]);
+        return result > 0;
+      }
+      return !isNaN(Number(value)) && Number(value) > 0;
+    },
+  },
+  {
+    name: 'LOG_LEVEL',
+    required: false,
+    description: 'Logging level (trace, debug, info, warn, error, fatal)',
+    validator: (value) =>
+      ['trace', 'debug', 'info', 'warn', 'error', 'fatal'].includes(value),
+  },
+  {
+    name: 'NODE_ENV',
+    required: false,
+    description: 'Node environment (development, production, test)',
+    validator: (value) => ['development', 'production', 'test'].includes(value),
+  },
+];
+
+export function validateEnvironment(): void {
+  const errors: string[] = [];
+  const warnings: string[] = [];
+
+  logger.info('Validating environment variables...');
+
+  for (const envVar of ENV_VARIABLES) {
+    const value = process.env[envVar.name];
+
+    if (envVar.required) {
+      if (!value || value.trim() === '') {
+        errors.push(
+          `Required environment variable ${envVar.name} is missing. ${envVar.description}`
+        );
+        continue;
+      }
+    }
+
+    if (value && envVar.validator) {
+      try {
+        if (!envVar.validator(value)) {
+          errors.push(
+            `Environment variable ${envVar.name} has invalid format. ${envVar.description}`
+          );
+        }
+      } catch (error) {
+        errors.push(
+          `Environment variable ${envVar.name} validation failed: ${String(error)}. ${envVar.description}`
+        );
+      }
+    }
+
+    if (!envVar.required && !value) {
+      warnings.push(
+        `Optional environment variable ${envVar.name} not set, using default. ${envVar.description}`
+      );
+    }
+  }
+
+  if (warnings.length > 0) {
+    logger.warn('Environment variable warnings:');
+    warnings.forEach((warning) => { logger.warn(`  - ${warning}`); });
+  }
+
+  if (errors.length > 0) {
+    logger.error('Environment variable validation failed:');
+    errors.forEach((error) => { logger.error(`  - ${error}`); });
+    logger.error('Please check your environment configuration and try again.');
+    process.exit(1);
+  }
+
+  logger.info('Environment variable validation completed successfully');
+}