fix: harden session error handling, normalizer null guard, extract redis-connection

- Wrap groupFetchAllParticipating in try/catch to prevent unhandled rejection on connect
- Catch errors from async onMessage/onGroups callbacks via Promise.resolve().catch
- Return null from normalizer when key.id is missing (prevents empty upsert key collision)
- Extract parseRedisUrl to redis-connection.ts to eliminate duplication

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

apps/worker/src/queues/ingest.processor.ts

@@ -1,10 +1,6 @@
 import { Worker } from 'bullmq';
 import { IngestJobData } from '@tower/types';
-
-function parseRedisUrl(url: string) {
-  const { hostname, port } = new URL(url);
-  return { host: hostname, port: parseInt(port || '6379', 10), maxRetriesPerRequest: null };
-}
+import { parseRedisUrl } from './redis-connection';
 
 export async function processIngestJob(job: IngestJobData, prisma: any): Promise<void> {
   await prisma.message.upsert({

apps/worker/src/queues/ingest.queue.ts

@@ -1,10 +1,6 @@
 import { Queue } from 'bullmq';
 import { IngestJobData } from '@tower/types';
-
-function parseRedisUrl(url: string) {
-  const { hostname, port } = new URL(url);
-  return { host: hostname, port: parseInt(port || '6379', 10), maxRetriesPerRequest: null };
-}
+import { parseRedisUrl } from './redis-connection';
 
 export function createIngestQueue(redisUrl: string): Queue<IngestJobData> {
   return new Queue<IngestJobData>('tower:ingest', { connection: parseRedisUrl(redisUrl) });

apps/worker/src/queues/redis-connection.ts

@@ -0,0 +1,4 @@
+export function parseRedisUrl(url: string) {
+  const { hostname, port } = new URL(url);
+  return { host: hostname, port: parseInt(port || '6379', 10), maxRetriesPerRequest: null };
+}