MSW + SSE

handler.ts

   sendMercure({
      ctx,
      action: '/activities/{id}',
      data,
    }).then()

msw.helpers.ts

export async function sendMercure<T>({
  ctx,
  action,
  data,
  sleep = 0,
  count,
}: {
  ctx: RestContext
  action: string
  data: T
  sleep?: number
  count?: number
}) {
  if (sleep)
    await new Promise(resolve => setTimeout(resolve, sleep))

  await ctx.fetch('http://localhost:3000/send-message', {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json', // Set the content type to JSON
      'Loop-Count': count ? `${count}` : '0',
    },
    body: JSON.stringify({ action: `/api${action}`, data }),
  })
}

sse.package.json

{
  "name": "sse",
  "version": "1.0.0",
  "description": "",
  "main": "index.js",
  "scripts": {
    "dev": "DEBUG=express:* nodemon index.ts"
  },
  "keywords": [],
  "devDependencies": {
    "@types/cors": "^2.8.15",
    "@types/express": "^4.17.20",
    "@typescript-eslint/eslint-plugin": "^6.4.0",
    "eslint": "^8.0.1",
    "eslint-config-standard-with-typescript": "^39.1.1",
    "eslint-plugin-import": "^2.25.2",
    "eslint-plugin-n": "^15.0.0 || ^16.0.0 ",
    "eslint-plugin-promise": "^6.0.0",
    "tsconfig": "*",
    "typescript": "*"
  },
  "author": "",
  "license": "ISC",
  "dependencies": {
    "@types/express": "^4.17.20",
    "better-sse": "^0.10.0",
    "cors": "^2.8.5",
    "express": "^4.18.2",
    "nodemon": "^3.0.1",
    "ts-node": "^10.9.1",
    "zod": "^3.22.4"
  }
}

sse.ts

import { createChannel, createSession } from 'better-sse'
import express, { type Request, type Response } from 'express' // Import express types // Import express types
import cors from 'cors'
import { z } from 'zod'

function getRandomDelaySeconds (min: number, max: number): number {
  return (Math.random() * (max - min) + min) * 1000
}

const app = express()
const port = 3000
const resources: Record<string, ReturnType<typeof createChannel>> = {}
app.use(cors())
app.use(express.json()) // Middleware to parse JSON
const querySchema = z.object({
  topic: z.string().refine((topic) => (topic !== '') && topic.includes('/api'), {
    message: 'Invalid topic format'
  })
})
// eslint-disable-next-line @typescript-eslint/no-misused-promises
app.get('/.well-known/mercure', async (req: Request, res: Response) => { // Add type annotations to req and res
  try {
    const { topic: rawTopic } = querySchema.parse(req.query)

    console.log('TOPICRAW', rawTopic)
    const topic = decodeURIComponent(rawTopic)
    console.log('TOPIC', topic)
    // eslint-disable-next-line @typescript-eslint/strict-boolean-expressions
    if (!resources[topic]) {
      resources[topic] = createChannel()
    }
    const session = await createSession(req, res) // Assuming createSession accepts Request and Response

    resources[topic].register(session)
  } catch (error: any) {
    return res.status(400).json(error)
  }
})

// Define a schema for the request body using Zod
const sendMessageSchema = z.object({
  action: z.string(), // Action as a string
  data: z.unknown() // Data as any
})

function broadcastAction (action: string, data: any): void {
  setTimeout(() => {
    resources[action].broadcast(data)
  }, getRandomDelaySeconds(0.2, 1.4))
}
// eslint-disable-next-line @typescript-eslint/no-misused-promises
app.post('/send-message', async (req: Request, res: Response) => {
  try {
    const { action: rawAction, data } = sendMessageSchema.parse(req.body)
    // get the customer  Loop-Count header
    const action = decodeURIComponent(rawAction)
    if (action.includes('{{anyId}}')) {
      const loopCountHeader = req.header('Loop-Count')
      const loopCount = loopCountHeader ? parseInt(loopCountHeader) : 79

      for (let i = 1; i <= loopCount; i++) {
        const modifiedAction = action.replace('{{anyId}}', i.toString())
        if (resources[modifiedAction]) {
          console.log('Broadcasting to topic', modifiedAction)
          broadcastAction(modifiedAction, data)
        } else {
          console.log('No resource found for topic', modifiedAction)
        }
      }
    } else if (resources[action]) {
      console.log('Broadcasting to topic', action)
      broadcastAction(action, data)
    } else {
      console.log('No resource found for topic', action)
    }
    return res.send('ok')
  } catch (error) {
    return res.status(400).json({ error: 'Invalid request format' })
  }
})
app.listen(port, () => {
  console.log(`Server is rundnindg on port ${port}`)
})