import { type SseTicketStore } from '@pipeline-builder/api-core';
import type { Response } from 'express';
import type { SSERelay } from './sse-relay.js';
/**
 * Event types for SSE logging
 */
export type SSEEventType = 'INFO' | 'WARN' | 'ERROR' | 'COMPLETED' | 'ROLLBACK' | 'MESSAGE';
/**
 * SSE payload structure
 */
export interface SSEPayload {
    ts: string;
    type: SSEEventType;
    message: string;
    data?: unknown;
}
/**
 * SSE client with connection tracking
 */
export interface SSEClient {
    id: string;
    res: Response;
    connectedAt: number;
    timeout: NodeJS.Timeout;
    /** Number of consecutive backpressure events (write returned false). */
    backpressureCount: number;
    /** Owning org id — populated when the request was authenticated. Used to
     *  decrement the per-org counter on disconnect / cleanup. */
    orgId?: string;
}
/**
 * SSE Manager configuration options
 */
export interface SSEManagerOptions {
    /** Maximum clients allowed per request ID (default: 10) */
    maxClientsPerRequest?: number;
    /** Client timeout in milliseconds (default: 30 minutes) */
    clientTimeoutMs?: number;
    /** Interval for cleanup checks in milliseconds (default: 5 minutes) */
    cleanupIntervalMs?: number;
    /**
     * Hard cap on total open connections per process. Defaults to 1000 from
     * `SSE_MAX_TOTAL_CLIENTS`. New connections beyond this are rejected at
     * `addClient()`. Tune up if your service serves > 1000 concurrent SSE
     * dashboards, but be aware Node.js fd limits dominate above ~5000.
     */
    maxTotalClients?: number;
    /**
     * Per-org ceiling on concurrent SSE streams. Defaults to 50 from
     * `SSE_MAX_CLIENTS_PER_ORG`. Protects against one noisy org consuming the
     * process-wide budget — without it, an authenticated user can fan out
     * thousands of streams (one per dashboard / log stream) and starve other
     * orgs of connection slots. The cap is enforced only on `addClient` calls
     * that carry an `orgId`; service-internal / anonymous SSE traffic skips it.
     */
    maxClientsPerOrg?: number;
    /**
     * Cap on live log-stream tickets across the process, for the DEFAULT
     * in-memory ticket store. Defaults to 1000 from `SSE_MAX_TOTAL_TICKETS`.
     * Ignored when `ticketStore` is injected (the store carries its own caps).
     */
    maxTotalTickets?: number;
    /**
     * Per-org cap on live log-stream tickets, for the DEFAULT in-memory ticket
     * store. Defaults to 10 from `SSE_MAX_TICKETS_PER_ORG`. Ignored when
     * `ticketStore` is injected.
     */
    maxTicketsPerOrg?: number;
    /** Ticket TTL in ms for the default store (default: `SSE_TICKET_TTL_MS` from api-core). */
    ticketTtlMs?: number;
    /**
     * Ticket + stream-ownership backend for the subject-bound log stream. Defaults
     * to an in-memory store (single replica only), built on first use. Inject an
     * api-core `createEnvSseTicketStore(...)` so mint/redeem and stream ownership
     * work across pods.
     */
    ticketStore?: SseTicketStore;
    /**
     * Whether this service streams per-request logs (`ctx.log` frames keyed by
     * requestId, the `/logs` channel). Default false: `ctx.log` then only writes
     * to the logger, so a service that never serves a log stream doesn't push
     * every log line through SSE (and over Redis).
     */
    logStream?: boolean;
    /**
     * TTL for a stream-ownership binding (default: `SSE_STREAM_OWNER_TTL_MS` env or
     * 1h). Long enough to outlive a build; the producer re-binds as needed.
     */
    streamOwnerTtlMs?: number;
    /**
     * Cross-pod fan-out bus for `send()`/`broadcast()`, wired at construction.
     * When omitted (and no `relayFactory` is enabled) delivery is LOCAL-ONLY.
     * Degrades to local-only during a Redis outage.
     */
    relay?: SSERelay;
    /**
     * Builds the relay on demand — see {@link SSEManager.enableRelay}. Lets a
     * service open a relay connection only when it actually has a streaming
     * channel (the log stream, or an org-keyed channel). Returning null (Redis
     * not configured) keeps local-only delivery.
     */
    relayFactory?: () => SSERelay | null;
}
/**
 * A build-log stream subject id: either a dashed UUID (api-server's `uuid()`)
 * or nginx's `$request_id` — 16 random bytes rendered as 32 hex chars with NO
 * dashes. Accept both by making the group separators optional; still strictly
 * 32 hex chars so it can't carry an injection / path-traversal payload into the
 * SSE subject. Shared by the ticket-mint route and the stream middleware so
 * both validate the subject identically.
 */
export declare const SSE_REQUEST_ID_RE: RegExp;
/** Result of {@link SSEManager.createTicket}. */
export type CreateTicketResult = {
    ok: true;
    ticket: string;
} | {
    ok: false;
    reason: 'org-limit' | 'capacity' | 'forbidden';
};
/**
 * SSE Manager statistics
 */
export interface SSEManagerStats {
    totalRequests: number;
    totalClients: number;
    oldestConnectionMs: number | null;
}
/**
 * SSE helper class with memory leak protection
 *
 * Features:
 * - Client limits per request ID
 * - Automatic timeout for idle connections
 * - Periodic cleanup of stale connections
 * - Connection statistics
 *
 * @example
 * ```typescript
 * const sseManager = new SSEManager({ maxClientsPerRequest: 5 });
 * app.get('/logs/:requestId', sseManager.middleware());
 *
 * // Send events
 * sseManager.send('request-123', 'INFO', 'Processing...');
 * ```
 */
export declare class SSEManager {
    private clients;
    /** Per-org open-client counters. Decremented on removeClient/cleanup so the
     *  map mirrors `clients[].orgId` counts at all times. Orgs reach zero are
     *  deleted to keep the map bounded. */
    private orgClientCounts;
    /** Injected ticket + stream-ownership backend, or the lazily built in-memory default. */
    private ticketStoreInstance?;
    /** True when this manager built its ticket store (so shutdown stops it). */
    private ownsTicketStore;
    private readonly maxClientsPerRequest;
    private readonly maxTotalClients;
    private readonly maxClientsPerOrg;
    private readonly maxTotalTickets;
    private readonly maxTicketsPerOrg;
    private readonly ticketTtlMs;
    private readonly streamOwnerTtlMs;
    private readonly clientTimeoutMs;
    private cleanupInterval;
    /** Cross-pod fan-out bus (undefined ⇒ local-only delivery). */
    private relay?;
    private readonly relayFactory?;
    /** Whether `ctx.log` frames are streamed (see {@link SSEManagerOptions.logStream}). */
    readonly logStreamEnabled: boolean;
    /** This manager instance's id — tags relayed frames so we ignore our own echo. */
    private readonly nodeId;
    constructor(options?: SSEManagerOptions);
    private attachRelay;
    /**
     * Turn on cross-pod delivery using the configured `relayFactory`. Idempotent;
     * a no-op when a relay is already wired or no factory was given, and when the
     * factory returns null (Redis not configured → local-only delivery).
     */
    enableRelay(): void;
    /** The ticket store, building the in-memory default on first use. */
    private get ticketStore();
    /**
     * Handle a frame received from another pod via the relay. We already delivered
     * our OWN frames to local clients before publishing, so ignore our echo. A
     * `send` re-emits to the subject's local clients; a `broadcast` to all of them.
     * Purely local — never re-publishes, so there is no fan-out loop.
     */
    private onRelayMessage;
    /** Total open connections across all requests. */
    private totalClients;
    /**
     * Publish the current live-connection count as a gauge so on-call can see SSE
     * saturation (approach to the per-process cap) on a dashboard — previously
     * these numbers lived only in `getStats()`/logs. Called after every add/remove/
     * close/cleanup. Cheap (O(R)); metric helpers never throw.
     */
    private updateActiveGauge;
    /** Current open-stream count for an org (0 if unseen). */
    getOrgClientCount(orgId: string): number;
    /**
     * Canonical form of a requestId for equality checks: dashes stripped and
     * lowercased, so the dashed (api-server `uuid()`) and undashed (nginx
     * `$request_id`) renderings of the same id compare equal. Callers are
     * expected to have already format-validated against {@link SSE_REQUEST_ID_RE}.
     */
    private static normalizeRequestId;
    /**
     * Canonical form of an orgId for ownership equality checks: trimmed and
     * lowercased. The stream PRODUCER (bindStreamOwner) and the ticket-minting
     * CONSUMER (createTicket) run in different services, so a casing / whitespace
     * drift between how each renders the same org would otherwise make the owner
     * comparison in {@link createTicket} false-`forbidden` the real owner.
     * Normalizing both sides — mirroring {@link normalizeRequestId} — closes that.
     */
    private static normalizeOrgId;
    /**
     * Record the org that OWNS a stream subject. The stream PRODUCER calls this
     * when it creates a build-log stream so that ticket minting can assert the
     * caller's org owns the subject (see {@link createTicket}). Backed by the
     * injected ticket store, so when a Redis store is wired the platform producer
     * (a different service sharing the same Redis) can bind ownership that this
     * service reads — closing the cross-tenant attach gap where any org could mint
     * a ticket for a guessed requestId.
     *
     * @param requestId - The stream subject (format-validated by the caller).
     * @param orgId - The owning org (normalized internally via normalizeOrgId).
     */
    bindStreamOwner(requestId: string, orgId: string): Promise<void>;
    /**
     * Mint a short-lived, single-use SSE ticket bound to `orgId` AND a specific
     * stream subject (`requestId`). Clients POST to obtain one (JWT-authenticated)
     * for the exact stream they intend to open, then open the EventSource with
     * `?ticket=<t>` so the JWT never lands in a query string / access log. The
     * ticket can then only be consumed to open that one subject's stream — see
     * {@link consumeTicket} — which is what enforces per-subject authorization.
     *
     * ORG-OWNERSHIP: if a stream owner has been bound for this subject (via
     * {@link bindStreamOwner}) and it is a DIFFERENT org, minting is refused
     * (`reason: 'forbidden'`) — an org cannot mint a ticket for another org's
     * stream even if it guesses the requestId. When no owner is bound (producer
     * wiring not yet present), minting falls back to binding the ticket to the
     * caller's own org, preserving current behavior.
     *
     * Bounded by the ticket store's live-ticket caps (total and per org). An
     * ownership lookup the store can't answer is refused as `capacity`.
     *
     * @param orgId - Owning org (normalized internally via normalizeOrgId).
     * @param requestId - The build-log stream subject this ticket authorizes.
     *   The caller must have format-validated it (see {@link SSE_REQUEST_ID_RE}).
     * @returns `{ ok: true, ticket }` on success, or `{ ok: false, reason }`
     *   where reason is `'org-limit'`, `'capacity'`, or `'forbidden'`.
     */
    createTicket(orgId: string, requestId: string): Promise<CreateTicketResult>;
    /**
     * Validate and CONSUME a ticket for a specific stream subject. Single-use:
     * the ticket is deleted whether or not it turns out to be valid, so a replay
     * of the same value always fails. Returns the bound org on success, or null
     * when the ticket is unknown / already-used / expired / OR was minted for a
     * DIFFERENT `requestId` than the one being opened.
     *
     * The subject check is the authorization fix: without it, a ticket minted for
     * one stream could be presented to open ANY stream the holder's org could
     * name, letting an authenticated org attach to another org's log stream by
     * guessing its requestId. The generic 401 the caller returns does not
     * distinguish "unknown ticket" from "wrong subject", so it leaks nothing
     * about which requestIds exist.
     *
     * @param ticketId - The opaque ticket value from `?ticket=`.
     * @param requestId - The stream subject from the URL path (`:requestId`),
     *   already format-validated by the caller.
     */
    consumeTicket(ticketId: string, requestId: string): Promise<{
        orgId: string;
    } | null>;
    /**
     * Adds a client to the SSE manager
     *
     * @param requestId - Unique request ID
     * @param res - Express Response object
     * @param orgId - Authenticated org id (optional). When set, enforces the
     *   per-org cap and the counter is decremented on disconnect/cleanup.
     * @param maxClientsForSubject - Per-subject cap override (default
     *   `maxClientsPerRequest`). An org-keyed channel's subject IS the org, so it
     *   must not inherit the small per-build-request cap — see {@link addOrgClient}.
     * @returns true if client was added, false if rejected (limit reached)
     */
    addClient(requestId: string, res: Response, orgId?: string, maxClientsForSubject?: number): boolean;
    /**
     * Attach a client to an ORG-keyed stream (the org is the subject). Counts
     * against the per-org cap, and the org — not the per-request cap — bounds how
     * many streams share the subject.
     */
    addOrgClient(orgId: string, res: Response): boolean;
    /** Drop one from the per-org counter. Idempotent: a counter at 0 stays at 0
     *  and the org key is removed from the map. */
    private decrementOrgCount;
    /**
     * Removes a client from the manager
     */
    private removeClient;
    /**
     * Sends a message to all SSE clients for a requestId
     *
     * @param requestId - Request ID
     * @param type - Event type
     * @param message - Message string
     * @param data - Optional additional data
     * @returns Number of clients the message was sent to
     */
    send(requestId: string, type: SSEEventType, message: string, data?: unknown): number;
    /**
     * Write a pre-built payload to this pod's LOCAL clients for `requestId`. Shared
     * by {@link send} (local half) and the relay re-emit path ({@link onRelayMessage}),
     * so a relayed frame reuses the producer's original `ts` rather than a new one.
     * Never relays — callers decide whether to publish.
     */
    private sendLocal;
    /**
     * Broadcast a message to all connected clients across all requests
     *
     * @param type - Event type
     * @param message - Message string
     * @param data - Optional additional data
     * @returns Total number of clients the message was sent to
     */
    broadcast(type: SSEEventType, message: string, data?: unknown): number;
    /** Write a pre-built payload to all LOCAL clients across every request. */
    private broadcastLocal;
    /** Fire-and-forget relay publish. No-op (local-only) when no relay is wired. */
    private publishRelay;
    /**
     * Close all clients for a specific request
     *
     * @param requestId - Request ID to close
     * @param finalMessage - Optional final message to send before closing
     */
    closeRequest(requestId: string, finalMessage?: string): void;
    /**
     * Get statistics about current connections
     */
    getStats(): SSEManagerStats;
    /**
     * Check if a request has any connected clients
     */
    hasClients(requestId: string): boolean;
    /**
     * Get the number of clients for a specific request
     */
    getClientCount(requestId: string): number;
    /**
     * Middleware to initialize a ticket-authenticated SSE connection.
     *
     * The stream is NOT open to the world: the caller must first exchange its
     * JWT for a short-lived single-use ticket (see {@link createTicket}, wired to
     * `POST /logs/ticket`) and pass it as `?ticket=<t>`. The middleware validates
     * + consumes the ticket, binds the connection to the ticket's org, and
     * enforces the per-org connection cap. A missing / invalid / expired /
     * already-used ticket is rejected with 401 before any SSE headers flush.
     *
     * @example
     * ```typescript
     * app.get('/logs/:requestId', sseManager.middleware());
     * ```
     */
    middleware(): (req: {
        params: {
            requestId: string;
        };
        query?: {
            ticket?: unknown;
        };
    }, res: Response) => Promise<void>;
    /**
     * Start periodic cleanup of stale connections
     */
    private startCleanupInterval;
    /**
     * Clean up stale connections using single-pass partition.
     * O(R × C) instead of O(R × C²), and avoids mutation-during-iteration.
     */
    private cleanup;
    /**
     * Shutdown the SSE manager and close all connections
     */
    shutdown(): void;
}