claude sonnet 4 refactor run

src/README.md

@@ -0,0 +1,103 @@
+# Backend Architecture
+
+This document describes the refactored backend architecture for the live-crawler application.
+
+## Directory Structure
+
+```
+src/
+├── config/           # Configuration management
+├── controllers/      # HTTP request handlers
+├── services/         # Business logic services
+├── routes/           # API route definitions
+├── types/           # TypeScript interfaces and types
+├── utils/           # Utility functions
+├── models/          # Data models (existing files)
+├── app.ts           # Express app factory
+└── index.ts         # Application entry point
+```
+
+## Architecture Overview
+
+The refactored backend follows a layered architecture pattern with clear separation of concerns:
+
+### 1. Configuration Layer (`config/`)
+- Centralizes all application configuration
+- Environment-specific settings
+- Type-safe configuration objects
+
+### 2. Controller Layer (`controllers/`)
+- Handles HTTP requests and responses
+- Input validation and error handling
+- Delegates business logic to services
+
+### 3. Service Layer (`services/`)
+- Contains business logic
+- Manages external dependencies (Streamr client)
+- Orchestrates data processing operations
+
+### 4. Route Layer (`routes/`)
+- Defines API endpoints
+- Maps HTTP routes to controller methods
+- Middleware configuration
+
+### 5. Utility Layer (`utils/`)
+- Reusable utility functions
+- Input validation helpers
+- Common operations
+
+### 6. Types Layer (`types/`)
+- TypeScript interfaces and types
+- Shared data structures
+- API contract definitions
+
+## Key Components
+
+### StreamrService
+Manages the Streamr client lifecycle and provides network node access.
+
+### TopologyService
+Handles topology crawling and data processing operations.
+
+### TopologyController
+Processes HTTP requests for topology endpoints with proper error handling.
+
+## Benefits of Refactoring
+
+1. **Separation of Concerns**: Each layer has a specific responsibility
+2. **Testability**: Services and controllers can be easily unit tested
+3. **Maintainability**: Code is organized and easier to understand
+4. **Scalability**: New features can be added without affecting existing code
+5. **Error Handling**: Centralized and consistent error handling
+6. **Type Safety**: Strong TypeScript typing throughout the application
+7. **Dependency Injection**: Services are injected, making the code more flexible
+
+## API Endpoints
+
+### GET /topology
+Retrieves network topology for a given stream.
+
+**Query Parameters:**
+- `streamId` (required): The stream identifier
+- `partition` (optional): The partition number (defaults to 0)
+
+**Response:**
+Returns an array of topology nodes with their neighbors and RTT information.
+
+### GET /health
+Health check endpoint that returns the application status.
+
+## Error Handling
+
+The application uses a centralized error handling approach:
+- Custom `ValidationError` for input validation errors
+- Proper HTTP status codes
+- Structured error logging
+- Graceful error responses
+
+## Graceful Shutdown
+
+The application implements graceful shutdown handling:
+- Listens for SIGTERM and SIGINT signals
+- Properly closes the Streamr client
+- Ensures all resources are cleaned up

src/Topology.ts

@@ -4,19 +4,7 @@ import { Multimap, numberToIpv4, StreamPartIDUtils } from '@streamr/utils'
 import { NormalizedNodeInfo } from './NetworkNodeFacade'
 import { NodeType } from "@streamr/trackerless-network/dist/generated/packages/dht/protos/DhtRpc"
 import { fetchLocationData } from "./fetchLocationData"
-
-export interface Node {
-    id: DhtAddress
-    ipAddress?: string
-    applicationVersion: string
-    websocketUrl?: string
-    nodeType: string
-    region?: number
-    streamPartNeighbors: Multimap<StreamPartID, DhtAddress>
-    controlLayerNeighborCount: number
-    allStreamPartitions: string[]
-    location?: any
-}
+import { Node } from './types'
 
 export class Topology {
     private nodes: Map<DhtAddress, Node> = new Map()

src/app.ts

@@ -0,0 +1,37 @@
+import express from 'express'
+import cors from 'cors'
+import path from 'path'
+import { Logger } from '@streamr/utils'
+import { createRoutes } from './routes'
+import { TopologyController } from './controllers/TopologyController'
+
+const logger = new Logger(module)
+
+export function createApp(topologyController: TopologyController): express.Application {
+    const app = express()
+
+    // Middleware
+    app.use(cors())
+    app.use(express.static(path.join(__dirname, '../public')))
+
+    // Routes
+    app.use('/', createRoutes(topologyController))
+
+    // Health check endpoint
+    app.get('/health', (_req, res) => {
+        res.json({ status: 'ok', timestamp: new Date().toISOString() })
+    })
+
+    // 404 handler
+    app.use('*', (_req, res) => {
+        res.status(404).json({ error: 'Not Found' })
+    })
+
+    // Error handler
+    app.use((err: Error, req: express.Request, res: express.Response, _next: express.NextFunction) => {
+        logger.error('Unhandled error', { err, url: req.url, method: req.method })
+        res.status(500).json({ error: 'Internal Server Error' })
+    })
+
+    return app
+}

src/config/index.ts

@@ -0,0 +1,11 @@
+export const config = {
+    port: process.env.PORT ?? 3000,
+    streamr: {
+        metrics: false
+    },
+    crawler: {
+        maxConcurrent: 5
+    }
+} as const
+
+export type Config = typeof config

src/controllers/TopologyController.ts

@@ -0,0 +1,48 @@
+import { Request, Response } from 'express'
+import { Logger } from '@streamr/utils'
+import { NetworkNodeFacade } from '../NetworkNodeFacade'
+import { TopologyService } from '../services/TopologyService'
+import { StreamrService } from '../services/StreamrService'
+import { validateStreamId, validatePartition, createStreamPartId, ValidationError } from '../utils/validation'
+import { TopologyQueryParams } from '../types'
+
+export class TopologyController {
+    private readonly logger = new Logger(module)
+    private readonly topologyService: TopologyService
+    private readonly streamrService: StreamrService
+
+    constructor(topologyService: TopologyService, streamrService: StreamrService) {
+        this.topologyService = topologyService
+        this.streamrService = streamrService
+    }
+
+    async getTopology(req: Request<object, object, object, TopologyQueryParams>, res: Response): Promise<void> {
+        try {
+            // Validate request parameters
+            const streamId = validateStreamId(req.query.streamId)
+            const partition = validatePartition(req.query.partition)
+            const streamPartId = createStreamPartId(streamId, partition)
+
+            // Get network node and fetch topology
+            const networkNode = await this.streamrService.getNetworkNode()
+            const localNode = new NetworkNodeFacade(networkNode)
+
+            const topology = await this.topologyService.getTopology(localNode, streamPartId)
+
+            res.json(topology)
+            this.logger.info('Sent topology as JSON response', { streamPartId })
+        } catch (err) {
+            this.handleError(err, res)
+        }
+    }
+
+    private handleError(err: unknown, res: Response): void {
+        if (err instanceof ValidationError) {
+            res.status(400).send(`Bad Request: ${err.message}`)
+            return
+        }
+
+        this.logger.warn('Encountered error while processing topology request', { err })
+        res.status(500).send('Internal Server Error')
+    }
+}

src/crawlTopology.ts

@@ -25,14 +25,14 @@ export const crawlTopology = async (
     localNode: NetworkNodeFacade,
     entryPoints: PeerDescriptor[],
     getNeighbors: (nodeInfo: NormalizedNodeInfo) => PeerDescriptor[],
-    runId: string
+    runId: string,
+    maxConcurrent: number = 5
 ): Promise<{ topology: Topology, rttMatrix: RttMatrix }> => {
     const startTime = Date.now()
     const nodeInfos: Map<DhtAddress, NormalizedNodeInfo> = new Map()
     const errorNodes: Set<DhtAddress> = new Set()
     const visitedNodes: Set<DhtAddress> = new Set() // used to avoid duplicate entries in queue
     const queue: PeerDescriptor[] = [...entryPoints]
-    const MAX_CONCURRENT = 5
     const activePromises: Set<Promise<void>> = new Set()
 
     const processNode = async (peerDescriptor: PeerDescriptor) => {
@@ -63,7 +63,7 @@ export const crawlTopology = async (
     }
 
     while (queue.length > 0 || activePromises.size > 0) {
-        while (queue.length > 0 && activePromises.size < MAX_CONCURRENT) {
+        while (queue.length > 0 && activePromises.size < maxConcurrent) {
             const peerDescriptor = queue.shift()!
             const promise = processNode(peerDescriptor).finally(() => activePromises.delete(promise))
             activePromises.add(promise)

src/fetchLocationData.ts

@@ -2,18 +2,7 @@ import { GeoIpLocator } from '@streamr/geoip-location'
 import { Logger } from '@streamr/utils'
 import iso3166Data from './iso-3166-data.json'
 import { CityResponse } from "mmdb-lib/lib/reader/response"
-
-interface LocationData {
-    city?: string
-    country?: string
-    countryCode?: string
-    loc?: string
-    hostname?: string
-    org?: string
-    postal?: string
-    timezone?: string
-    subRegion?: string
-}
+import { LocationData } from './types'
 
 const logger = new Logger(module)