// ❌ Anti-Pattern — no cancellation propagation
// file: src/routes/streaming-broken.ts

app.post('/stream', async (req, res) => {
  // Stream starts. User navigates away after 2 seconds.
  // req is closed. res is closed. But the generator keeps running...
  for await (const token of openai.stream({ messages })) {
    res.write(token)  // Throws silently on closed socket, but loop continues
  }
  // The above loop runs for another 30 seconds generating 2,000 tokens
  // that no one reads, costing real money per token
})

// ✅ Backpressure-safe streaming pipeline
// file: src/gateway/BackpressureStream.ts

/**
 * Wraps an AsyncGenerator in a WHATWG ReadableStream with proper backpressure.
 * highWaterMark controls the internal queue depth:
 * - Too low (1): excessive start/stop cycles, overhead
 * - Too high (100): defeats backpressure purpose
 * - Sweet spot (8-16): balances throughput vs memory
 *
 * IMPORTANT: In object mode (string chunks), highWaterMark is the NUMBER OF OBJECTS,
 * not bytes. A value of 8 means the queue holds 8 string chunks, not 8 bytes.
 * Use CountQueuingStrategy for explicit object-mode semantics:
 *   new ReadableStream(src, new CountQueuingStrategy({ highWaterMark: 8 }))
 */
export function generatorToReadableStream<T>(
  generator: AsyncGenerator<T>,
  highWaterMark = 8,
): ReadableStream<T> {
  return new ReadableStream<T>(
    {
      async pull(controller) {
        const { value, done } = await generator.next()
        if (done) {
          controller.close()
        } else {
          controller.enqueue(value)
        }
      },
      cancel(reason) {
        // Propagate cancellation back into the generator
        generator.return?.(reason)
      },
    },
    new CountQueuingStrategy({ highWaterMark }),
  )
}

/**
 * Token serialisation TransformStream: converts typed WorkspaceActions
 * into SSE-formatted text chunks ready for the HTTP response sink.
 */
export function createSseTransform(): TransformStream<WorkspaceAction, string> {
  let eventId = 0
  return new TransformStream<WorkspaceAction, string>({
    transform(action, controller) {
      eventId++
      controller.enqueue(`id: ${eventId}\ndata: ${JSON.stringify(action)}\n\n`)
    },
    flush(controller) {
      controller.enqueue('data: [DONE]\n\n')
    },
  })
}

// ✅ Full production streaming route
// file: src/routes/chat-stream.ts

import { pipeline } from 'node:stream/promises'
import { Readable } from 'node:stream'

export async function chatStreamRoute(req: FastifyRequest, reply: FastifyReply) {
  const body = ChatRequestSchema.parse(req.body)

  // ── Cancellation chain setup ──────────────────────────────────────────────
  const controller = new AbortController()

  // Wire client disconnect to abort signal — the critical billing-leak prevention
  req.raw.on('close', () => {
    if (!controller.signal.aborted) {
      controller.abort(new Error('Client disconnected'))
    }
  })

  // ── Response headers ──────────────────────────────────────────────────────
  reply.raw.writeHead(200, {
    'Content-Type': 'text/event-stream',
    'Cache-Control': 'no-cache, no-transform',
    'Connection': 'keep-alive',
    'X-Accel-Buffering': 'no',
  })
  reply.raw.flushHeaders()

  // ── Streaming pipeline ────────────────────────────────────────────────────
  const gateway = req.diScope.resolve<AiGateway>('aiGateway')
  const actionGenerator = gateway.streamResponse(body, controller.signal)

  // Wrap generator → ReadableStream with backpressure
  const readableStream = generatorToReadableStream(actionGenerator, 8)

  // Add SSE serialisation transform
  const { readable, writable } = createSseTransform()
  readableStream.pipeTo(writable).catch(() => {/* absorbed — abort signal handles teardown */})

  // Convert WHATWG ReadableStream → Node.js Readable for pipe to res.raw
  const nodeReadable = Readable.fromWeb(readable as any)

  try {
    await pipeline(nodeReadable, reply.raw, { signal: controller.signal })
  } catch (error) {
    if (controller.signal.aborted) {
      // Normal: client disconnected or server cancelled
      return
    }
    // Unexpected: log and surface
    req.log.error(error, 'Streaming pipeline error')
  }
}

// ✅ Composite abort signal — cancelled by ANY of the three conditions
// file: src/gateway/CompositeCancel.ts

export function buildCancellationSignal(
  clientSignal: AbortSignal,       // From req.raw.on('close')
  timeoutMs: number,               // Max generation time budget
  orgLimitSignal: AbortSignal,     // Org-level rate limit kill switch
): AbortSignal {
  const timeoutSignal = AbortSignal.timeout(timeoutMs)

  // AbortSignal.any() — available Node.js 20+
  // Fires when ANY of the provided signals aborts
  return AbortSignal.any([clientSignal, timeoutSignal, orgLimitSignal])
}

// Usage in route:
const signal = buildCancellationSignal(
  req.raw.socket as unknown as AbortSignal, // wrapped
  30_000,     // 30 second max per stream
  org.signal, // org-level emergency stop
)

// ✅ Streaming metrics — emit these to OpenTelemetry (Part 23)
// file: src/gateway/StreamMetrics.ts

export class StreamingMonitor {
  private activeStreams = new Map<string, StreamSession>()

  startSession(sessionId: string): void {
    this.activeStreams.set(sessionId, {
      startedAt: Date.now(),
      tokensEmitted: 0,
      bytesBuffered: 0,
    })
  }

  recordToken(sessionId: string, bufferDepth: number): void {
    const session = this.activeStreams.get(sessionId)
    if (!session) return
    session.tokensEmitted++
    session.bytesBuffered = bufferDepth

    // Alert if buffer depth exceeds threshold — client not keeping up
    if (bufferDepth > 1024 * 64) { // 64KB buffer depth threshold
      console.warn(`[StreamingMonitor] High buffer depth: ${sessionId} @ ${bufferDepth} bytes`)
    }
  }

  endSession(sessionId: string, reason: 'complete' | 'aborted' | 'error'): void {
    const session = this.activeStreams.get(sessionId)
    if (!session) return
    const duration = Date.now() - session.startedAt
    // Emit to OTEL span (wired in Part 23)
    console.info(`[Stream] ${sessionId}: ${session.tokensEmitted} tokens, ${duration}ms, reason: ${reason}`)
    this.activeStreams.delete(sessionId)
  }
}