sse-resilience

name	sse-resilience
description	Redis-backed SSE stream management with stream registry, heartbeat monitoring, completion store for terminal events, and automatic orphan cleanup via background guardian process.
license	MIT
compatibility	TypeScript/JavaScript
metadata	{"category":"api","time":"7h","source":"drift-masterguide"}

SSE Stream Resilience

Redis-backed stream management with heartbeat monitoring and completion recovery.

When to Use This Skill

• SSE streams can fail silently (client disconnects mid-stream)
• Completion events get lost and users never see results
• Need visibility into stream health
• Want to prevent resource leaks from abandoned streams

Core Concepts

The solution provides:

• Stream registry (track all active streams in Redis)
• Heartbeat monitoring (detect orphaned streams)
• Completion store (persist terminal events for recovery)
• Stream guardian (background cleanup process)

Client ←→ SSE Endpoint ←→ Stream Registry (Redis)
                              ↓
                    Completion Store (Redis)
                              ↓
                    Stream Guardian (Background)

Implementation

TypeScript

// lib/sse/types.ts
export enum StreamState {
  ACTIVE = 'active',
  COMPLETED = 'completed',
  FAILED = 'failed',
  ORPHANED = 'orphaned',
}

export interface StreamMetadata {
  streamId: string;
  streamType: string;
  userId: string;
  startedAt: Date;
  lastHeartbeat: Date;
  state: StreamState;
  metadata: Record<string, unknown>;
}

export interface CompletionData {
  streamId: string;
  terminalEventType: string;
  terminalEventData: Record<string, unknown>;
  completedAt: Date;
}

// lib/sse/stream-registry.ts
const STREAM_KEY_PREFIX = 'sse:stream:';
const ACTIVE_STREAMS_KEY = 'sse:active';
const  = ;        
  = ;     

   {
  () {}

   (: ): <> {
     streamKey = ;

     ( ..(streamKey)) {
       ;
    }

     pipeline = ..();

    pipeline.(streamKey, {
      : metadata.,
      : metadata.,
      : metadata.,
      : metadata..(),
      : metadata..(),
      : metadata.,
      : .(metadata.),
    });
    pipeline.(streamKey, );
    pipeline.(, metadata..(), metadata.);

     pipeline.();
     ;
  }

   (: ): <> {
     streamKey = ;
     now =  ();

     (! ..(streamKey)) {
       ;
    }

     pipeline = ..();
    pipeline.(streamKey, , now.());
    pipeline.(, now.(), streamId);
     pipeline.();

     ;
  }

   (: ): <> {
     streamKey = ;
     userId =  ..(streamKey, );

     (!userId)  ;

     pipeline = ..();
    pipeline.(streamKey);
    pipeline.(, streamId);
     pipeline.();

     ;
  }

   (thresholdSeconds = ): <[]> {
     cutoff = .() - (thresholdSeconds * );
     staleIds =  ..(, , cutoff);

     : [] = [];
     ( streamId  staleIds) {
       stream =  .(streamId);
       (stream && stream. === .) {
        streams.(stream);
      }
    }

     streams;
  }

   (: , : ): <> {
     streamKey = ;
     (! ..(streamKey))  ;
     ..(streamKey, , state);
     ;
  }
}


  = ;
  = ;  

   {
  () {}

   (: ): <> {
     key = ;

     ..(key, {
      : data.,
      : data.,
      : .(data.),
      : data..(),
    });
     ..(key, );
  }

   (: ): < | > {
     key = ;
     data =  ..(key);

     (!data.)  ;

     {
      : data.,
      : data.,
      : .(data. || ),
      :  (data.),
    };
  }
}


   {
   : . |  = ;

  () {}

  ():  {
     (.) ;

    . = (
       .(),
      .
    );
  }

  ():  {
     (.) {
      (.);
      . = ;
    }
  }

    (): <> {
     {
       staleStreams =  ..();

       ( stream  staleStreams) {
         .(stream);
      }
    }  (err) {
      .(, err);
    }
  }

    (: ): <> {
    .();
     ..(stream., .);
  }
}

SSE Endpoint

// app/api/stream/[streamId]/route.ts
export async function GET(req: Request, { params }: { params: { streamId: string } }) {
  const userId = req.headers.get('x-user-id')!;
  const streamId = params.streamId;

  // Check for existing completion (recovery)
  const existingCompletion = await completionStore.getCompletion(streamId);
  if (existingCompletion) {
    return new Response(
      `data: ${JSON.stringify({
        type: existingCompletion.terminalEventType,
        data: existingCompletion.terminalEventData,
      })}\n\n`,
      { headers: { 'Content-Type': 'text/event-stream' } }
    );
  }

  // Register new stream
  await registry.register({
    streamId,
    streamType: 'generation',
    userId,
    startedAt: new Date(),
    lastHeartbeat: new Date(),
    state: StreamState.ACTIVE,
    metadata: {},
  });

  const encoder = new TextEncoder();
   : .;

   stream =  ({
    () {
      controller.(
        encoder.()
      );

      
      heartbeatInterval = ( () => {
         {
           registry.(streamId);
          controller.(encoder.());
        }  {}
      }, );
    },

    () {
      (heartbeatInterval);
      registry.(streamId);
    },
  });

    (stream, {
    : {
      : ,
      : ,
      : streamId,
    },
  });
}

Client-Side Recovery

function useResilientSSE(streamId: string) {
  const [status, setStatus] = useState<'connecting' | 'connected' | 'completed' | 'error'>('connecting');
  const [data, setData] = useState<unknown>(null);
  const reconnectAttempts = useRef(0);

  useEffect(() => {
    let eventSource: EventSource | null = null;

    const connect = async () => {
      // First, check for existing completion
      try {
        const recovery = await fetch(`/api/stream/${streamId}/recover`);
        const result = await recovery.json();

        if (result.status === 'completed') {
          setStatus('completed');
          setData(result.terminalEventData);
          return;
        }
      } catch {}

      // Connect to SSE
      eventSource = new EventSource();

      eventSource. =  {
         parsed = .(event.);
        
         (parsed. ===  || parsed. === ) {
          ();
          (parsed.);
          eventSource?.();
        }  {
          (parsed);
        }
      };

      eventSource. =  {
        eventSource?.();
        
         (reconnectAttempts. < ) {
          reconnectAttempts.++;
          (connect,  * reconnectAttempts.);
        }  {
          ();
        }
      };
    };

    ();
      eventSource?.();
  }, [streamId]);

   { status, data };
}

Best Practices

1. Heartbeat every 15 seconds - Keeps stream alive and detects orphans
2. Store completions for recovery - 5 minute window for client reconnection
3. Background guardian process - Clean up orphaned streams automatically
4. Client-side reconnection - Retry with exponential backoff
5. Check for completion on connect - Recover missed terminal events

Common Mistakes

• No heartbeat mechanism (can't detect orphaned streams)
• Not storing completion data (lost terminal events)
• Missing recovery endpoint (clients can't recover)
• No background cleanup (resource leaks)
• Forgetting to unregister on clean disconnect

Related Patterns

• websocket-management - WebSocket alternative
• graceful-shutdown - Drain streams on shutdown
• checkpoint-resume - Track stream progress