[{"_1":2,"_176":-5,"_177":-5},"loaderData",{"_3":4,"_92":93},"root",{"_5":6,"_35":36,"_65":66},"ctx",{"_7":8,"_23":-5,"_24":25,"_26":27,"_28":20,"_29":20,"_30":20,"_31":32,"_33":20,"_34":20},"auth",{"_9":-5,"_10":-5,"_11":-5,"_12":-5,"_13":-5,"_14":-5,"_15":-5,"_16":-5,"_17":-5,"_18":-5,"_19":20,"_21":22},"sessionClaims","sessionId","sessionStatus","userId","actor","orgId","orgRole","orgSlug","orgPermissions","factorVerificationAge","isAuthenticated",false,"tokenType","session_token","user","preferencesKeybindings","default","preferencesTheme","light","preferencesFixers","preferencesCopilot","preferencesSelectionAi","notificationCount",0,"flagDenols","isTeamOrgMember","ENV",{"_37":38,"_39":40,"_41":42,"_43":44,"_45":46,"_47":48,"_49":50,"_51":52,"_53":54,"_55":56,"_57":58,"_59":60,"_61":62,"_63":64},"REACT_APP_SENTRY_DSN","https://586020d86f05489db63a554690acf1e9@o1398359.ingest.sentry.io/6724499","VAL_RUNTIME_NAME","prod","RENDER_GIT_COMMIT","5f8f00a127b1349c32de089cedf4e10bc7855654","RENDER_GIT_BRANCH","main","CLERK_PUBLISHABLE_KEY","pk_live_Y2xlcmsudmFsLnRvd24k","POSTHOG_API_TOKEN","phc_nmoyuAkStEZuUp69rYlg1DZgrIt19Gx0Y4Hwf8tGgJP","POSTHOG_API_HOST","https://wart.val.town","WEB_API_TEMPLATE","https://{val}.web.val.run","APP_URL","https://www.val.town","MODULE_URL","https://esm.town","VANITY_WEB_API_TEMPLATE","https://{label}.val.run/","DENOLS_URL","https://lsp.val.town","NODE_ENV","production","API_URL","https://api.val.town","clerkState",{"_67":68},"__internal_clerk_state",{"_69":70,"_71":46,"_72":73,"_74":73,"_75":20,"_76":73,"_77":73,"_78":73,"_79":73,"_80":81,"_90":20,"_91":20},"__clerk_ssr_state",{"_21":22,"_9":-5,"_10":-5,"_11":-5,"_12":-5,"_13":-5,"_14":-5,"_15":-5,"_16":-5,"_17":-5,"_18":-5,"_19":20},"__publishableKey","__proxyUrl","","__domain","__isSatellite","__signInUrl","__signUpUrl","__afterSignInUrl","__afterSignUpUrl","__clerk_debug",{"_82":20,"_19":20,"_83":73,"_84":85,"_86":73,"_87":46,"_88":20,"_89":73},"isSignedIn","proxyUrl","reason","session-token-and-uat-missing","message","publishableKey","isSatellite","domain","__telemetryDisabled","__telemetryDebug","routes/embed.x.$handle.$name.$",{"_94":95,"_96":97,"_98":99,"_100":101,"_151":152,"_173":174,"_175":-5},"handle","pinsar","name","feistyAmberRhinoceros","path","backend/processing/streamProcessor.ts","project",{"_102":103,"_96":97,"_104":105,"_106":-5,"_107":20,"_108":109,"_110":-5,"_111":-5,"_12":112,"_113":114,"_115":116,"_117":-5,"_118":119,"_120":32,"_14":121,"_122":123,"_124":123,"_125":126,"_127":126,"_128":126,"_129":123,"_130":123,"_23":131,"_134":135,"_137":138,"_150":20},"id","2149c16a-277d-11f0-b46e-569c3dd06744","createdAt",["D",1746207883134],"deletedAt","legacyEnvVars","privacy","public","description","imageUrl","851f6760-f7af-11ef-9322-0fd6f1c36e70","pinnedFileId","214e03d8-277d-11f0-b46e-569c3dd06744","pinPreview",true,"pinMigrated","screenshotVersion",19,"likeCount","094b6920-183c-466c-985a-fe1516b5d507","userPerm","r","valPerm","blobPerm","rw","emailPerm","sqlitePerm","projectPerm","telemetryPerm",{"_94":95,"_132":133},"profileImageUrl","https://img.clerk.com/eyJ0eXBlIjoicHJveHkiLCJzcmMiOiJodHRwczovL2ltYWdlcy5jbGVyay5kZXYvb2F1dGhfZ2l0aHViL2ltZ18ydG1LWVR3QzZpNUhHQkNmbzF5VlRvOGZweUQifQ","org",{"_94":95,"_136":133,"_12":112},"avatar","branch",{"_139":140,"_96":44,"_102":141,"_142":103,"_143":-5,"_144":-5,"_145":-5,"_146":-5,"_147":-5,"_148":97,"_149":95},"version",27,"214a9ba8-277d-11f0-b46e-569c3dd06744","projectId","forkedBranchId","forkedBranchVersion","forkedProjectPrivacy","forkedProjectId","forkedProjectOrgId","projectName","userHandle","isOwned","file",{"_102":153,"_96":154,"_155":156,"_157":158,"_159":141,"_160":161,"_162":116,"_104":163,"_139":32,"_164":165,"_166":167,"_168":20,"_169":-5,"_98":170},"809539b2-3a04-11f0-99a9-9e149126039e","streamProcessor.ts","fileId","80953a84-3a04-11f0-99a9-9e149126039e","parentId","4166da8d-efe9-4872-828b-925ab2d06100","branchId","branchVersion",14,"latest",["D",1748245145998],"fileType","script","content","import { EventStorage, AggregateStorage, Event, Aggregate } from \"../storage/database.ts\";\nimport { MetricsCollector } from \"../monitoring/metricsCollector.ts\";\n\nexport interface ProcessingWindow {\n start: number;\n end: number;\n events: Event[];\n}\n\nexport interface ProcessingResult {\n windowStart: number;\n windowEnd: number;\n eventsProcessed: number;\n aggregates: Aggregate;\n processingTime: number;\n}\n\nexport class StreamProcessor {\n private eventStorage: EventStorage;\n private aggregateStorage: AggregateStorage;\n private metricsCollector: MetricsCollector;\n private isRunning = false;\n private processingInstances = 1;\n private windowSizeMs = 10000; // 10 second windows\n private lastProcessedTime = 0;\n private faultToleranceEnabled = true;\n private retryAttempts = 3;\n\n constructor(metricsCollector: MetricsCollector) {\n this.eventStorage = new EventStorage();\n this.aggregateStorage = new AggregateStorage();\n this.metricsCollector = metricsCollector;\n }\n\n start(): void {\n if (this.isRunning) {\n console.log(\"Stream processor already running\");\n return;\n }\n\n this.isRunning = true;\n this.lastProcessedTime = Date.now() - this.windowSizeMs;\n \n console.log(\"Starting stream processor with fault tolerance\");\n \n // Start processing loop\n this.processLoop();\n \n // Start background aggregation\n this.startAggregationLoop();\n }\n\n stop(): void {\n this.isRunning = false;\n console.log(\"Stream processor stopped\");\n }\n\n scale(instances: number): void {\n this.processingInstances = Math.max(1, Math.min(instances, 10));\n console.log(`Scaled stream processor to ${this.processingInstances} instances`);\n \n // Update metrics\n this.metricsCollector.setGauge('processing_instances', this.processingInstances);\n }\n\n async reset(): Promise {\n this.lastProcessedTime = Date.now() - this.windowSizeMs;\n console.log(\"Stream processor state reset\");\n }\n\n private async processLoop(): Promise {\n while (this.isRunning) {\n try {\n await this.processNextWindow();\n \n // Dynamic sleep based on load\n const sleepTime = this.calculateOptimalSleepTime();\n await this.sleep(sleepTime);\n \n } catch (error) {\n console.error(\"Error in processing loop:\", error);\n \n if (this.faultToleranceEnabled) {\n await this.handleProcessingError(error);\n } else {\n throw error;\n }\n }\n }\n }\n\n private async processNextWindow(): Promise {\n const windowStart = this.lastProcessedTime;\n const windowEnd = windowStart + this.windowSizeMs;\n const currentTime = Date.now();\n \n // Don't process future windows\n if (windowEnd > currentTime - 1000) {\n return null;\n }\n\n const startTime = performance.now();\n \n try {\n // Fetch events for this window\n const events = await this.eventStorage.getEventsByTimeRange(windowStart, windowEnd);\n \n if (events.length === 0) {\n this.lastProcessedTime = windowEnd;\n return null;\n }\n\n // Process events with fault tolerance\n const result = await this.processWindowWithRetry({\n start: windowStart,\n end: windowEnd,\n events\n });\n\n // Update last processed time\n this.lastProcessedTime = windowEnd;\n \n const processingTime = performance.now() - startTime;\n \n // Update metrics\n await this.metricsCollector.recordWindowProcessed(\n events.length,\n processingTime\n );\n\n console.log(`Processed window: ${events.length} events in ${processingTime.toFixed(2)}ms`);\n \n return {\n windowStart,\n windowEnd,\n eventsProcessed: events.length,\n aggregates: result,\n processingTime\n };\n \n } catch (error) {\n console.error(`Failed to process window ${windowStart}-${windowEnd}:`, error);\n throw error;\n }\n }\n\n private async processWindowWithRetry(window: ProcessingWindow): Promise {\n let lastError: Error | null = null;\n \n for (let attempt = 1; attempt <= this.retryAttempts; attempt++) {\n try {\n return await this.processWindow(window);\n } catch (error) {\n lastError = error as Error;\n console.warn(`Processing attempt ${attempt} failed:`, error);\n \n if (attempt < this.retryAttempts) {\n // Exponential backoff\n await this.sleep(Math.pow(2, attempt) * 100);\n }\n }\n }\n \n throw lastError;\n }\n\n private async processWindow(window: ProcessingWindow): Promise {\n const { start, end, events } = window;\n \n // Calculate processing times\n const processingTimes = events\n .map(e => e.processing_time_ms)\n .filter(t => t !== null && t !== undefined) as number[];\n \n // Calculate aggregates\n const aggregate: Aggregate = {\n window_start: start,\n window_end: end,\n event_count: events.length,\n avg_processing_time: this.calculateAverage(processingTimes),\n min_processing_time: Math.min(...processingTimes, 0),\n max_processing_time: Math.max(...processingTimes, 0),\n p95_processing_time: this.calculatePercentile(processingTimes, 95),\n p99_processing_time: this.calculatePercentile(processingTimes, 99),\n error_count: 0, // Would be calculated from error events\n throughput_per_second: events.length / ((end - start) / 1000)\n };\n\n // Store aggregate\n await this.aggregateStorage.storeAggregate(aggregate);\n \n // Process by event type\n const eventsByType = this.groupEventsByType(events);\n \n for (const [eventType, typeEvents] of Object.entries(eventsByType)) {\n const typeProcessingTimes = typeEvents\n .map(e => e.processing_time_ms)\n .filter(t => t !== null && t !== undefined) as number[];\n \n const typeAggregate: Aggregate = {\n window_start: start,\n window_end: end,\n event_type: eventType,\n event_count: typeEvents.length,\n avg_processing_time: this.calculateAverage(typeProcessingTimes),\n min_processing_time: Math.min(...typeProcessingTimes, 0),\n max_processing_time: Math.max(...typeProcessingTimes, 0),\n p95_processing_time: this.calculatePercentile(typeProcessingTimes, 95),\n p99_processing_time: this.calculatePercentile(typeProcessingTimes, 99),\n error_count: 0,\n throughput_per_second: typeEvents.length / ((end - start) / 1000)\n };\n \n await this.aggregateStorage.storeAggregate(typeAggregate);\n }\n\n return aggregate;\n }\n\n private async startAggregationLoop(): Promise {\n // Run aggregation every 30 seconds\n const aggregationInterval = 30000;\n \n const runAggregation = async () => {\n if (!this.isRunning) return;\n \n try {\n await this.runRealTimeAggregation();\n } catch (error) {\n console.error(\"Aggregation error:\", error);\n }\n \n setTimeout(runAggregation, aggregationInterval);\n };\n \n setTimeout(runAggregation, aggregationInterval);\n }\n\n private async runRealTimeAggregation(): Promise {\n const now = Date.now();\n const fiveMinutesAgo = now - (5 * 60 * 1000);\n \n // Get recent aggregates\n const recentAggregates = await this.aggregateStorage.getAggregateHistory(1);\n \n if (recentAggregates.length === 0) return;\n \n // Calculate rolling averages\n const totalEvents = recentAggregates.reduce((sum, agg) => sum + agg.event_count, 0);\n const avgThroughput = recentAggregates.reduce((sum, agg) => sum + agg.throughput_per_second, 0) / recentAggregates.length;\n const avgLatency = recentAggregates.reduce((sum, agg) => sum + agg.avg_processing_time, 0) / recentAggregates.length;\n \n // Update real-time metrics\n await this.metricsCollector.setGauge('total_events_5min', totalEvents);\n await this.metricsCollector.setGauge('avg_throughput_per_second', avgThroughput);\n await this.metricsCollector.setGauge('avg_processing_latency_ms', avgLatency);\n \n // Check for anomalies and auto-scaling triggers\n await this.checkForAnomalies(avgThroughput, avgLatency);\n }\n\n private async checkForAnomalies(throughput: number, latency: number): Promise {\n // Auto-scaling logic\n if (throughput > 5000 && this.processingInstances < 5) {\n console.log(\"High throughput detected, auto-scaling up\");\n this.scale(this.processingInstances + 1);\n } else if (throughput < 1000 && this.processingInstances > 1) {\n console.log(\"Low throughput detected, auto-scaling down\");\n this.scale(this.processingInstances - 1);\n }\n \n // Latency alerts\n if (latency > 100) {\n console.warn(`High latency detected: ${latency.toFixed(2)}ms`);\n await this.metricsCollector.incrementCounter('high_latency_alerts');\n }\n \n // Throughput alerts\n if (throughput > 10000) {\n console.warn(`Very high throughput: ${throughput.toFixed(2)} events/sec`);\n await this.metricsCollector.incrementCounter('high_throughput_alerts');\n }\n }\n\n private async handleProcessingError(error: Error): Promise {\n console.error(\"Processing error handled:\", error);\n \n await this.metricsCollector.incrementCounter('processing_errors_total');\n \n // Implement circuit breaker pattern\n const errorRate = await this.metricsCollector.getErrorRate();\n if (errorRate > 0.1) { // 10% error rate\n console.warn(\"High error rate detected, implementing backoff\");\n await this.sleep(5000); // 5 second backoff\n }\n }\n\n private groupEventsByType(events: Event[]): Record {\n const grouped: Record = {};\n \n for (const event of events) {\n if (!grouped[event.type]) {\n grouped[event.type] = [];\n }\n grouped[event.type].push(event);\n }\n \n return grouped;\n }\n\n private calculateAverage(numbers: number[]): number {\n if (numbers.length === 0) return 0;\n return numbers.reduce((sum, n) => sum + n, 0) / numbers.length;\n }\n\n private calculatePercentile(numbers: number[], percentile: number): number {\n if (numbers.length === 0) return 0;\n \n const sorted = [...numbers].sort((a, b) => a - b);\n const index = Math.ceil((percentile / 100) * sorted.length) - 1;\n return sorted[Math.max(0, index)];\n }\n\n private calculateOptimalSleepTime(): number {\n // Dynamic sleep based on processing instances and load\n const baseSleep = 1000; // 1 second base\n const scaleFactor = Math.max(0.1, 1 / this.processingInstances);\n return Math.floor(baseSleep * scaleFactor);\n }\n\n private sleep(ms: number): Promise {\n return new Promise(resolve => setTimeout(resolve, ms));\n }\n}","deleted","label",[171,172,154],"backend","processing","permalink","https://www.val.town/x/pinsar/feistyAmberRhinoceros/code/backend/processing/streamProcessor.ts","extension","actionData","errors"]