SSE Package Documentation

import { Hono } from 'hono';
import { streamSSE } from 'hono/streaming';
import { CacheHub } from '@countcachula/sse';

const app = new Hono();
const hub = new CacheHub();

app.get('/api/cache-events', (c) => {
  return streamSSE(c, async (stream) => {
    // Register connection
    hub.addConnection(stream);

    // Send initial connection event
    await stream.writeSSE({
      event: 'connected',
      data: JSON.stringify({ timestamp: Date.now() })
    });

    // Clean up on disconnect
    c.req.raw.signal.addEventListener('abort', () => {
      hub.removeConnection(stream);
      console.log('Client disconnected');
    });

    // Keep connection alive with heartbeat
    while (!c.req.raw.signal.aborted) {
      await new Promise(resolve => setTimeout(resolve, 30000));

      // Send heartbeat
      await stream.writeSSE({
        event: 'heartbeat',
        data: 'ping'
      });
    }
  });
});

import express from 'express';
import { CacheHub } from '@countcachula/sse';

const app = express();
const hub = new CacheHub();

app.get('/api/cache-events', (req, res) => {
  // Set SSE headers
  res.writeHead(200, {
    'Content-Type': 'text/event-stream',
    'Cache-Control': 'no-cache',
    'Connection': 'keep-alive'
  });

  // Create stream wrapper
  const stream = {
    writeSSE: async ({ event, data }) => {
      res.write(`event: ${event}\n`);
      res.write(`data: ${data}\n\n`);
    }
  };

  hub.addConnection(stream);

  // Send initial event
  stream.writeSSE({
    event: 'connected',
    data: 'connected'
  });

  // Clean up on disconnect
  req.on('close', () => {
    hub.removeConnection(stream);
  });
});

// When an entity changes, invalidate all related caches
app.patch('/api/users/:id', async (c) => {
  const userId = c.req.param('id');
  const updates = await c.req.json();

  await updateUser(userId, updates);

  // Invalidate specific user and any lists containing them
  await hub.invalidate([
    `user:${userId}`,              // User detail
    'users-list',                  // Main list
    `team:${updates.teamId}:members`, // Team member list
    'stats:users'                  // User statistics
  ]);

  return c.json({ success: true });
});

// Invalidate parent and all children
app.delete('/api/projects/:id', async (c) => {
  const projectId = c.req.param('id');

  await deleteProject(projectId);

  await hub.invalidate([
    `project:${projectId}`,
    `project:${projectId}:tasks`,
    `project:${projectId}:members`,
    `project:${projectId}:files`,
    'projects-list'
  ]);

  return c.json({ success: true });
});

// Periodically invalidate volatile data
setInterval(async () => {
  await hub.invalidate([
    'stats:realtime',
    'dashboard:metrics',
    'active-users'
  ]);
}, 60000); // Every minute

app.get('/api/posts', async (c) => {
  const posts = await getPosts();

  // After sending list, hint about individual posts
  if (posts.length > 0) {
    // Preload first 5 posts
    const urls = posts
      .slice(0, 5)
      .map(p => `/api/posts/${p.id}`);

    await hub.preloadHint(urls);
  }

  return c.json(posts);
});

app.get('/api/user/profile', async (c) => {
  const profile = await getUserProfile(c.userId);

  // Preload likely next navigation
  await hub.preloadHint([
    '/api/user/settings',     // Common next step
    '/api/user/notifications', // Often checked together
    `/api/users/${profile.id}/posts` // User's content
  ]);

  return c.json(profile);
});

app.get('/api/items', async (c) => {
  const page = parseInt(c.req.query('page') || '1');
  const items = await getItems(page);

  // Preload next page
  if (items.hasMore) {
    await hub.preloadHint([
      `/api/items?page=${page + 1}`
    ]);
  }

  return c.json(items);
});

app.post('/api/items/batch-update', async (c) => {
  const updates = await c.req.json(); // Array of updates

  const invalidationTags = [];
  const preloadUrls = [];

  for (const update of updates) {
    await updateItem(update.id, update.data);
    invalidationTags.push(`item:${update.id}`);
    preloadUrls.push(`/api/items/${update.id}`);
  }

  // Single invalidation for all changes
  invalidationTags.push('items-list');
  await hub.invalidate(invalidationTags);

  // Preload all updated items
  await hub.preloadHint(preloadUrls);

  return c.json({ updated: updates.length });
});

app.patch('/api/posts/:id', async (c) => {
  const id = c.req.param('id');
  const updates = await c.req.json();
  const oldPost = await getPost(id);

  await updatePost(id, updates);

  const tags = [`post:${id}`];

  // Only invalidate list if visibility changed
  if (updates.status !== oldPost.status) {
    tags.push('posts-list');
    tags.push(`posts:${updates.status}`);
  }

  // Only invalidate author's posts if author changed
  if (updates.authorId !== oldPost.authorId) {
    tags.push(`author:${oldPost.authorId}:posts`);
    tags.push(`author:${updates.authorId}:posts`);
  }

  await hub.invalidate(tags);

  return c.json({ success: true });
});

class DebouncedCacheHub extends CacheHub {
  constructor() {
    super();
    this.pendingInvalidations = new Set();
    this.debounceTimer = null;
  }

  async invalidate(tags: string[]) {
    tags.forEach(tag => this.pendingInvalidations.add(tag));

    if (this.debounceTimer) {
      clearTimeout(this.debounceTimer);
    }

    this.debounceTimer = setTimeout(async () => {
      const tags = Array.from(this.pendingInvalidations);
      this.pendingInvalidations.clear();
      await super.invalidate(tags);
    }, 100); // Batch invalidations within 100ms
  }
}

class SelectiveCacheHub extends CacheHub {
  constructor() {
    super();
    this.subscriptions = new Map(); // connection -> interests
  }

  addConnection(stream, interests = ['*']) {
    super.addConnection(stream);
    this.subscriptions.set(stream, new Set(interests));
  }

  async invalidate(tags: string[]) {
    for (const [connection, interests] of this.subscriptions) {
      // Only send if connection is interested
      const relevantTags = tags.filter(tag => {
        return interests.has('*') ||
               interests.has(tag) ||
               Array.from(interests).some(i => tag.startsWith(i));
      });

      if (relevantTags.length > 0) {
        await connection.writeSSE({
          event: 'invalidate',
          data: JSON.stringify(relevantTags)
        });
      }
    }
  }
}