Back to Blog
Real-Time Updates with Server-Sent Events and RTK Query Cache Invalidation
• 12 min read

Real-Time Updates with Server-Sent Events and RTK Query Cache Invalidation

Building a scalable real-time notification system from NestJS to React using SSE and RTK Query

by Jonathan Beurel

When building modern SaaS applications, keeping the UI synchronized with backend state is crucial. Users expect to see updates immediately when data changes—whether it's a credit balance update, a new transaction, or a team member joining.

In this article, I'll share how I built a production-ready real-time update system using Server-Sent Events (SSE) combined with RTK Query's cache invalidation mechanism. This architecture allows me to push updates from my NestJS backend through a Next.js proxy to my React frontend, automatically refreshing the UI without polling.

The Challenge: Keeping UI State Fresh

Imagine a credit-based billing system where users can:

  • Purchase credits through Stripe
  • Consume credits for API calls
  • Receive automatic top-ups asynchronously when their balance falls below a threshold
  • View real-time transaction history and updated balance

Traditional approaches include:

  1. Polling: Client requests data every N seconds

    • Simple but inefficient
    • Wastes bandwidth
    • Delayed updates

  2. WebSockets: Bi-directional real-time communication

    • More complex to implement
    • Requires connection state management on both ends
    • Overkill when you only need server→client updates

  3. Server-Sent Events (SSE): Uni-directional server→client updates

    • Built on standard HTTP
    • Automatic reconnection
    • Perfect for read-only real-time updates
    • Works through proxies and load balancers

I chose SSE because it provides the right balance of simplicity and power for my use case.

Architecture Overview

My system consists of three main layers:

sequenceDiagram
    participant Client as React Client
    participant Cache as RTK Query Cache
    participant BE as Next + Nest Backend
    participant Stripe
 
    BE->>Cache: Get and Cache<br/>Wallet data<br/>(30 credits)
    Cache->>Client: Display the Wallet
 
 
    Client->>BE: Consume credits: POST /credits/consume
    BE->>BE: Deduct<br/>(20 credits)
    Note over BE: Emit credit.consumed event
 
    par Response returned
        BE->>Cache: Update Wallet data
        Cache->>Client: Display the Wallet<br/>(10 credits)
    and Check threshold
        Note over BE: Listen credit.consumed event
        BE->>BE: Check balance (10 credits) < threshold (20 credits)
        BE->>Stripe: Create payment intent
        BE->>Cache: Update Transaction list
        Cache->>Client: Display<br/>Pending Transaction
    and Async Stripe Payment
        Note over BE: Listen Stripe webhook event
        Stripe-->>BE: Payment success webhook
        BE->>BE: Add 100 credits to wallet
        BE->>Cache: SSE<br/>tags_invalidated(['Credit'])
        Cache->>BE: Auto-refetch<br/>GET /credits/wallet
        BE->>Cache: Fresh wallet data
        Cache->>Client: Display new Wallet<br/>(110 credits)
    end

Let's break down each layer in detail.

Layer 1: NestJS Backend - Event Emission

The EventsService

At the heart of my SSE system is the EventsService, which manages Server-Sent Events streams for each workspace:

@Injectable()
export class EventsService implements OnModuleDestroy {
  private readonly eventSubjects = new Map<string, Subject<any>>();
 
  // Get or create an SSE subject for a specific workspace
  getWorkspaceEventStream(workspaceId: string): Subject<any> {
    if (!this.eventSubjects.has(workspaceId)) {
      const subject = new Subject<any>();
      this.eventSubjects.set(workspaceId, subject);
 
      // Clean up when no more subscribers
      subject.subscribe({
        complete: () => {
          this.eventSubjects.delete(workspaceId);
        },
      });
    }
 
    return this.eventSubjects.get(workspaceId)!;
  }
 
  // Emit a tags invalidation event to trigger RTK Query cache invalidation
  emitTagsInvalidated(workspaceId: string, tags: string[]) {
    const subject = this.eventSubjects.get(workspaceId);
    if (subject && !subject.closed) {
      subject.next({
        type: 'tags_invalidated',
        data: {
          tags,
          timestamp: new Date().toISOString(),
        },
      });
    }
  }
}

Key design decisions:

  1. Workspace-level streams: Each workspace gets its own RxJS Subject, ensuring isolation between tenants
  2. Automatic cleanup: When the last subscriber disconnects, I clean up resources to prevent memory leaks
  3. Type safety: Events have a type field for client-side discrimination

The EventsController

The controller exposes the SSE endpoint:

@Controller()
@UseGuards(ServerAuthGuard)
export class EventsController {
  constructor(private readonly eventsService: EventsService) {}
 
  @Sse('events')
  events(@CurrentWorkspace() workspace: Workspace): Observable<MessageEvent> {
    return this.eventsService.getWorkspaceEventStream(workspace.id).pipe(
      // Send immediate connection event to flush headers
      startWith({
        type: 'connection',
        data: { timestamp: new Date().toISOString() },
      }),
      map(
        (data) =>
          ({
            data: JSON.stringify(data),
            type: data.type,
          }) as MessageEvent
      )
    );
  }
}

Important points:

  • Authentication: Uses ServerAuthGuard to ensure only authenticated users can access events
  • Workspace context: The @CurrentWorkspace() decorator extracts the workspace from the request headers
  • Immediate connection event: I use startWith() to send an initial event that flushes headers immediately—crucial for establishing the SSE connection through proxies

Emitting Events from Business Logic

When business events occur (like credit top-up), I emit SSE events:

@Injectable()
export class CreditTopupListener {
  constructor(
    private readonly creditService: CreditService,
    private readonly stripeService: StripeService,
    private readonly eventsService: EventsService
  ) {}
 
  @OnEvent('credit.consumed')
  async handleCreditConsumed(event: { transactionId: string; walletId: string }) {
    const wallet = await this.creditService.getCreditWallet(event.walletId);
 
    // Check if automatic topup should be triggered
    if (wallet.balance <= wallet.threshold) {
      await this.stripeService.executeTopupPayment(wallet);
      this.eventsService.emitTagsInvalidated(wallet.workspaceId, ['Credit']);
    }
  }
}

The key is emitTagsInvalidated(['Credit']) — this tells all connected clients to invalidate any RTK Query cache entries tagged with 'Credit'.

Layer 2: React Client - SSE Connection Management

On the client side, I create a custom hook to manage the SSE connection:

// Note: This is a simplified version of the actual implementation for clarity in this blog post
export function useSSE() {
  const dispatch = useAppDispatch();
  const currentWorkspace = useAppSelector(getCurrentWorkspace);
  const [isConnected, setIsConnected] = useState(false);
 
  useEffect(() => {
    const url = `/api/events?workspaceId=${encodeURIComponent(currentWorkspace)}`;
    const eventSource = new EventSource(url, { withCredentials: true });
 
    eventSource.onopen = () => setIsConnected(true);
    eventSource.onerror = () => setIsConnected(false);
 
    eventSource.addEventListener('tags_invalidated', (event) => {
      // THIS IS THE MAGIC: Invalidate RTK Query cache tags
      dispatch(api.util.invalidateTags(event.data.tags));
    });
 
    return () => {
      eventSource.close();
      setIsConnected(false);
    };
  }, [currentWorkspace, dispatch]);
 
  return { isConnected };
}

Key feature

  • Workspace isolation: Reconnects when switching workspaces
  • connection state: Returns the connection state of the SSE connection

Layer 3: RTK Query - Cache Invalidation

RTK Query's tag-based cache invalidation is what makes this architecture elegant. I define my API endpoints with tags:

export const creditApi = api.injectEndpoints({
  endpoints: (builder) => ({
    getCreditWallet: builder.query<CreditWallet, void>({
      query: () => ({ url: '/credits/wallet', method: 'GET' }),
      providesTags: ['Credit'], // ← This query provides the 'Credit' tag
    }),
    getTransactions: builder.query<CreditTransaction[], void>({
      query: () => ({ url: '/credits/transactions', method: 'GET' }),
      providesTags: ['Credit'], // ← This query also provides the 'Credit' tag
    }),
    consumeCredits: builder.mutation<CreditTransaction, ConsumeCreditsRequest>({
      query: (body) => ({ url: '/credits/consume', method: 'POST', body }),
      invalidatesTags: ['Credit'], // ← This mutation invalidates 'Credit' tag
    }),
  }),
});
 
export const { useGetCreditWalletQuery, useGetTransactionsQuery, useConsumeCreditsMutation } = creditApi;

When the SSE event calls dispatch(api.util.invalidateTags(['Credit'])), RTK Query:

  1. Marks all cache entries with the 'Credit' tag as invalid
  2. Automatically refetches any currently subscribed queries (active components)
  3. Updates all components using those queries

This means zero manual refetching code in components!

Putting It All Together

Here's a real component using this system:

export function CreditBlock() {
  // Establish SSE connection
  const { isConnected } = useSSE();
 
  // RTK Query hooks (will auto-refetch on SSE invalidation)
  const { data: wallet } = useGetCreditWalletQuery();
  const { data: transactions } = useGetTransactionsQuery();
  const [consumeCredits, { isLoading }] = useConsumeCreditsMutation();
 
  const handleConsume = async () => {
    await consumeCredits({
      eventType: 'DEMO_CONSUMPTION',
      creditCost: 10,
    }).unwrap();
 
    toast.success('Credits consumed successfully');
  };
 
  return (
    <div>
      {/* SSE connection indicator */}
      <div className={isConnected ? 'bg-green-500' : 'bg-gray-400'} />
 
      {/* Wallet balance - auto-updates via SSE */}
      <div>Balance: {wallet.balance} credits</div>
 
      {/* Transaction history - auto-updates via SSE */}
      {transactions.map((tx) => (
        <div key={tx.id}>{tx.type}: {tx.delta}</div>
      ))}
 
      <button onClick={handleConsume} disabled={isLoading}>
        Consume Credits
      </button>
    </div>
  );
}

The beauty of this approach:

  • Component code is simple and declarative
  • No manual refetching or polling
  • No WebSocket complexity
  • Real-time updates work automatically
  • Connection state is easily accessible

The Complete Flow

Let's trace a complete user action through all three parallel branches:

Initial State

  • User starts with 30 credits in their wallet
  • Threshold is set to 20 credits

1. User Clicks "Consume Credits"

Request: Component calls consumeCredits() mutation with 20 credits

NestJS processes:

  • Deducts 20 credits from wallet (30 → 10 credits)
  • Emits credit.consumed event
  • Three things happen in parallel:

Branch 1: Immediate Response (~200ms)

What happens:

  • NestJS returns 200 OK to the client
  • RTK Query mutation updates the cache optimistically
  • UI immediately reflects the deduction

User sees:

Balance: 10 credits ✓

Branch 2: Threshold Check & Payment Intent (~500ms)

What happens:

  • Event listener catches credit.consumed event
  • Checks balance (10) < threshold (20) ✓
  • Creates Stripe Payment Intent
  • Creates a PENDING transaction in the database
  • RTK Query cache is updated

User sees:

Balance: 10 credits
Transactions:
  ⏳ Top-up (pending) - Processing...

Branch 3: Async Webhook Processing (~2-5 seconds)

What happens:

  • Stripe processes the payment (asynchronously)
  • Stripe webhook calls NestJS: payment_intent.succeeded
  • NestJS adds 100 credits to the wallet (10 → 110)
  • Updates transaction status: PENDING → COMPLETED
  • Emits SSE: tags_invalidated(['Credit'])
  • Client receives SSE event
  • RTK Query automatically refetches wallet and transactions

User sees:

Balance: 110 credits ✓
Transactions:
  ✅ Top-up +100 credits - Completed
  📉 Consumption -20 credits

Timeline Summary

t=0ms       User clicks button
t=200ms     ✓ UI shows 10 credits (immediate feedback)
t=500ms     ⏳ UI shows pending transaction
t=2-5s      ✅ UI shows 110 credits (webhook completed)

The beauty: Users get instant feedback, see the pending state, and eventually see the final result—all without writing manual refetch logic!

Production Considerations

1. Exponential Backoff

Network hiccups happen. Exponential backoff prevents overwhelming the server during reconnection:

const delay = 1000 * Math.pow(2, reconnectAttempts.current);
// 1s, 2s, 4s, 8s, 16s

2. Tab Visibility Handling

Mobile browsers aggressively close connections when tabs are backgrounded. Reconnecting on visibility change ensures a fresh connection:

const handleVisibilityChange = () => {
  if (document.visibilityState === 'visible') {
    if (eventSourceRef.current?.readyState === EventSource.CLOSED) {
      connect();
    }
  }
};

3. Memory Management

Clean up resources when workspaces disconnect:

subject.subscribe({
  complete: () => {
    this.eventSubjects.delete(workspaceId);
    this.connectionCounts.delete(workspaceId);
  },
});

4. Scalability

For horizontal scaling with multiple NestJS instances:

  • Use Redis Pub/Sub to broadcast events across instances
  • Each instance subscribes to a workspace channel
  • Events published to Redis are received by all instances
  • Each instance emits SSE events to its connected clients
// Publisher
await this.redis.publish(`workspace:${workspaceId}:events`, JSON.stringify({ type: 'tags_invalidated', data: { tags } }));
 
// Subscriber (in each instance)
await this.redis.subscribe(`workspace:${workspaceId}:events`);
this.redis.on('message', (channel, message) => {
  const event = JSON.parse(message);
  this.eventsService.emitToLocalClients(workspaceId, event);
});

Conclusion

By combining Server-Sent Events with RTK Query's tag-based cache invalidation, I've built a real-time update system that's:

  • Simple: Components just use RTK Query hooks with minimal code
  • Automatic: No manual refetching logic needed
  • Type-safe: Full TypeScript support end-to-end
  • Scalable: Works with horizontal scaling via Redis
  • Resilient: Automatic reconnection with backoff
  • Efficient: Only refetch what's needed via tag-based invalidation
  • Secure: Authentication handled by Next.js proxy
  • Observable: Easy to debug with event logging
  • Maintainable: Clear separation of concerns

The key insight is that you don't need to stream data over SSE—you only need to stream cache invalidation signals. RTK Query handles fetching the actual data using your existing REST endpoints.

If you're building a real-time SaaS application with Next.js and NestJS, I highly recommend this approach. The combination of SSE's simplicity and RTK Query's intelligent caching creates an elegant solution to a common problem.

Want to see the full implementation? Check out the Dopamine Starter Kit for a production-ready Next.js + NestJS starter with this SSE architecture built-in.