import {createHash, randomUUID} from 'node:crypto';
import type {Socket} from 'socket.io';
import type {BotEventName, ForwardAck, ForwardErrorCode, ForwardSessionStatus, JoinRoomReady,
    OpenSessionCommand, RelayClaims, RelayEventPayloads, RelayLimits, RelayMethod, RelayRequest, SessionScope} from '../types.ts';
import {RelayError, fail, is_record, is_uuid, success, error_ack, validate_open, validate_grant,
    validate_request, validate_event, validate_info_update, validate_node_ack} from '../utils/relay_validation.ts';
import {verify_token} from '../utils/token_utils.ts';

type Timer = ReturnType<typeof setTimeout>;
type Result = ForwardAck<unknown>;
interface DedupeEntry {fingerprint: string; expires_at: number; result?: Result; completed?: boolean; non_replayable: boolean; local?: boolean}
interface Grant {
    id: string; user_id: number; permissions: Set<RelayMethod>; expires_at: number;
    timer?: Timer; web?: Binding; dedupe: Map<string, DedupeEntry>;
}
interface Session {
    scope: SessionScope; lease_expires_at: number; timer?: Timer; reclaim?: Timer;
    node?: Binding; generation: number; ready: boolean; had_web: boolean; first_ready: boolean;
    grants: Map<string, Grant>; revoked_grants: Map<string, number>; webs: Set<Binding>; pending: Set<Pending>;
    stream?: string; seq?: number; retired?: Set<string>;
}
interface Connection {
    socket: Socket; binding?: Binding; busy: boolean; auth_timer?: Timer;
    cancel_join?: (error: RelayError) => void; joining_session?: Session; joining_grant?: Grant;
    requests: number; request_window: number;
    logs?: number; log_window?: number;
    disconnect: () => void;
}
interface Binding {
    connection: Connection; session: Session; claims: RelayClaims; grant?: Grant;
    expires_at: number; timer?: Timer; generation: number;
    subscriptions: Set<BotEventName>; pending: Set<Pending>;
}
interface Pending {
    id: string; binding: Binding; node: Binding; generation: number;
    session: Session; timer?: Timer; entry: DedupeEntry; cancel_transport?: () => void;
    finish: (result: Result) => void;
}
export interface SessionStoreOptions {
    forward_id: number; key: string; limits: RelayLimits;
    on_status: (status: ForwardSessionStatus) => void;
    on_error?: (error: unknown) => void;
}

function timer(callback: () => void, delay: number): Timer {
    const handle = setTimeout(callback, Math.max(1, Math.min(delay, 2_147_483_647)));
    handle.unref();
    return handle;
}
function same_scope(a: SessionScope, b: SessionScope): boolean {
    return a.forward_id === b.forward_id && a.forward_session_id === b.forward_session_id
        && a.node_session_id === b.node_session_id && a.node_id === b.node_id && a.bot_id === b.bot_id;
}
function same_identity(a: RelayClaims, b: RelayClaims): boolean {
    return same_scope(a, b) && a.role === b.role && a.grant_id === b.grant_id && a.user_id === b.user_id;
}
function canonical(value: unknown): string {
    if (Array.isArray(value)) return `[${value.map(canonical).join(',')}]`;
    if (is_record(value)) return `{${Object.keys(value).sort().map(key => `${JSON.stringify(key)}:${canonical(value[key])}`).join(',')}}`;
    return JSON.stringify(value);
}

export class RelaySessionStore {
    private readonly sessions = new Map<string, Session>();
    private readonly revoked_sessions = new Map<string, number>();
    private readonly connections = new Map<Socket, Connection>();
    private readonly pending = new Map<string, Pending>();
    private closed = false;
    private control = true;
    readonly limits: RelayLimits;
    private readonly options: SessionStoreOptions;

    constructor(options: SessionStoreOptions) {this.options = options; this.limits = options.limits;}
    get control_available(): boolean {return this.control;}
    get size(): number {return this.sessions.size;}
    set_control_available(available: boolean): void {if (!this.closed) this.control = available;}

    private report(session: Session, state: ForwardSessionStatus['state'], reason: string): void {
        try {
            this.options.on_status({...session.scope, node_ready: session.ready, state, reason, reported_at: Date.now()});
        } catch (error) {this.report_error(error);}
    }
    private report_error(error: unknown): void {
        try {this.options.on_error?.(error);} catch { /* Observer failures cannot retain relay resources. */ }
    }
    private active(session: Session): void {
        if (this.closed || this.sessions.get(session.scope.forward_session_id) !== session)
            fail('SESSION_STALE', 'Session is no longer active');
        if (session.lease_expires_at <= Date.now()) {
            this.close_session(session, 'lease_expired', 'TOKEN_EXPIRED');
            fail('TOKEN_EXPIRED', 'Session lease expired');
        }
    }
    private session(id: string): Session {
        const session = this.sessions.get(id);
        if (!session) fail('SESSION_STALE', 'Session is not registered');
        this.active(session);
        return session;
    }
    private control_required(): void {
        if (this.closed || !this.control) fail('RELAY_UNAVAILABLE', 'Control channel is unavailable', true);
    }
    private prune_revoked(entries: Map<string, number>): void {
        for (const [id, expires_at] of entries) if (expires_at <= Date.now()) entries.delete(id);
    }
    private tombstone_expiry(): number {return Date.now() + this.limits.max_token_age_ms + this.limits.clock_skew_ms;}
    open(command: unknown): OpenSessionCommand {
        this.control_required();
        validate_open(command, this.options.forward_id, this.limits);
        const value = command;
        this.prune_revoked(this.revoked_sessions);
        if (this.revoked_sessions.has(value.forward_session_id)) fail('SESSION_STALE', 'Session ID was revoked');
        const now = Date.now();
        if (value.lease_expires_at <= now || value.lease_expires_at > now + this.limits.max_lease_ms)
            fail('INVALID_REQUEST', 'Session lease is outside the allowed range');
        let session = this.sessions.get(value.forward_session_id);
        if (session && session.lease_expires_at <= now) {
            this.close_session(session, 'lease_expired', 'TOKEN_EXPIRED');
            fail('SESSION_STALE', 'Expired session ID cannot be reused');
        }
        if (session) {
            if (!same_scope(session.scope, value)) fail('SESSION_STALE', 'Session scope is immutable');
            session.lease_expires_at = Math.max(session.lease_expires_at, value.lease_expires_at);
        } else {
            for (const other of this.sessions.values()) {
                if (other.lease_expires_at <= now) this.close_session(other, 'lease_expired', 'TOKEN_EXPIRED');
                else if (other.scope.node_id === value.node_id && other.scope.bot_id === value.bot_id)
                    fail('SESSION_STALE', 'Revoke the existing Node/Bot session before opening another');
            }
            if (this.sessions.size + this.revoked_sessions.size >= this.limits.max_sessions) fail('RATE_LIMITED', 'Session capacity reached', true);
            const {forward_id, forward_session_id, node_session_id, node_id, bot_id} = value;
            session = {scope: {forward_id, forward_session_id, node_session_id, node_id, bot_id},
                lease_expires_at: value.lease_expires_at, generation: 0, ready: false, had_web: false,
                first_ready: false, grants: new Map(), revoked_grants: new Map(), webs: new Set(), pending: new Set()};
            this.sessions.set(value.forward_session_id, session);
            this.report(session, 'open', 'session_opened');
        }
        clearTimeout(session.timer);
        const owned = session;
        session.timer = timer(() => {
            if (this.sessions.get(value.forward_session_id) === owned)
                this.close_session(owned, 'lease_expired', 'TOKEN_EXPIRED');
        }, session.lease_expires_at - Date.now());
        return {...session.scope, lease_expires_at: session.lease_expires_at};
    }
    authorize(command: unknown) {
        this.control_required();
        validate_grant(command, this.limits);
        const value = command;
        const session = this.session(value.forward_session_id);
        this.prune_revoked(session.revoked_grants);
        if (session.revoked_grants.has(value.grant_id)) fail('FORBIDDEN', 'Grant ID was revoked');
        const now = Date.now();
        if (value.expires_at <= now || value.expires_at > now + this.limits.max_lease_ms
            || value.expires_at > session.lease_expires_at) fail('INVALID_REQUEST', 'Grant lease is outside the session lease');
        let grant = session.grants.get(value.grant_id);
        if (grant && grant.expires_at <= now) {
            this.close_grant(session, grant, 'grant_expired', 'TOKEN_EXPIRED');
            fail('FORBIDDEN', 'Expired grant ID cannot be reused');
        }
        if (grant) {
            if (grant.user_id !== value.user_id || grant.permissions.size !== value.permissions.length
                || value.permissions.some(permission => !grant!.permissions.has(permission)))
                fail('FORBIDDEN', 'Grant identity and permissions are immutable');
            grant.expires_at = Math.max(grant.expires_at, value.expires_at);
        } else {
            if (session.grants.size + session.revoked_grants.size >= this.limits.max_grants_per_session) fail('RATE_LIMITED', 'Grant capacity reached', true);
            grant = {id: value.grant_id, user_id: value.user_id, permissions: new Set(value.permissions),
                expires_at: value.expires_at, dedupe: new Map()};
            session.grants.set(grant.id, grant);
        }
        clearTimeout(grant.timer);
        const owned = grant;
        grant.timer = timer(() => {
            if (session.grants.get(owned.id) === owned) this.close_grant(session, owned, 'grant_expired', 'TOKEN_EXPIRED');
        }, grant.expires_at - Date.now());
        return {...value, expires_at: grant.expires_at};
    }
    private revoke_command(command: unknown, grant: boolean): {forward_session_id: string; reason: string; grant_id?: string} {
        if (!is_record(command) || !is_uuid(command.forward_session_id) || typeof command.reason !== 'string'
            || command.reason.trim().length < 1 || Buffer.byteLength(command.reason) > 256
            || (grant && !is_uuid(command.grant_id))
            || Object.keys(command).some(key => !['forward_session_id', 'reason', ...(grant ? ['grant_id'] : [])].includes(key)))
            fail('INVALID_REQUEST', 'Invalid revoke command');
        return command as {forward_session_id: string; reason: string; grant_id?: string};
    }
    revoke_session(command: unknown) {
        const value = this.revoke_command(command, false);
        const session = this.sessions.get(value.forward_session_id);
        if (session) this.close_session(session, value.reason, 'SESSION_STALE');
        else {
            this.prune_revoked(this.revoked_sessions);
            if (!this.revoked_sessions.has(value.forward_session_id) && this.sessions.size + this.revoked_sessions.size >= this.limits.max_sessions)
                fail('RATE_LIMITED', 'Revocation capacity reached', true);
            this.revoked_sessions.set(value.forward_session_id, this.tombstone_expiry());
        }
        return {forward_session_id: value.forward_session_id, revoked: true};
    }
    revoke_grant(command: unknown) {
        const value = this.revoke_command(command, true);
        const session = this.sessions.get(value.forward_session_id);
        const grant = session?.grants.get(value.grant_id!);
        if (grant) this.close_grant(session!, grant, value.reason, 'FORBIDDEN');
        else if (session) {
            this.prune_revoked(session.revoked_grants);
            if (!session.revoked_grants.has(value.grant_id!) && session.grants.size + session.revoked_grants.size >= this.limits.max_grants_per_session)
                fail('RATE_LIMITED', 'Grant revocation capacity reached', true);
            session.revoked_grants.set(value.grant_id!, this.tombstone_expiry());
        }
        return {forward_session_id: value.forward_session_id, grant_id: value.grant_id!, revoked: true};
    }

    attach(socket: Socket): boolean {
        if (this.closed || this.connections.size >= this.limits.max_connections) {
            socket.emit('forward_session_closed', {reason: 'connection_capacity', code: 'RATE_LIMITED'});
            socket.disconnect(true);
            return false;
        }
        const connection: Connection = {socket, busy: false, requests: 0, request_window: Date.now(),
            disconnect: () => this.detach(socket)};
        this.connections.set(socket, connection);
        connection.auth_timer = timer(() => this.drop(connection, 'authentication_timeout', 'UNAUTHORIZED'), this.limits.auth_timeout_ms);
        socket.on('disconnect', connection.disconnect);
        return true;
    }
    detach(socket: Socket): void {
        const connection = this.connections.get(socket);
        if (!connection) return;
        this.connections.delete(socket);
        connection.cancel_join?.(new RelayError('CANCELLED', 'Connection closed while joining'));
        clearTimeout(connection.auth_timer);
        socket.off('disconnect', connection.disconnect);
        const binding = connection.binding;
        if (!binding) return;
        connection.binding = undefined;
        clearTimeout(binding.timer);
        for (const pending of [...binding.pending]) pending.finish(error_ack(new RelayError('CANCELLED', 'Web connection closed')));
        const session = binding.session;
        if (binding.claims.role === 'node') {
            if (session.node !== binding) return;
            session.node = undefined;
            session.ready = false;
            for (const pending of [...session.pending]) pending.finish(error_ack(new RelayError('NODE_OFFLINE', 'Node disconnected', true)));
            this.status(session, 'node_disconnected');
        } else {
            if (binding.grant?.web === binding) binding.grant.web = undefined;
            session.webs.delete(binding);
            if (this.sessions.get(session.scope.forward_session_id) === session && session.webs.size === 0)
                this.reclaim(session, this.limits.idle_timeout_ms, 'last_web_left');
        }
    }
    private drop(connection: Connection, reason: string, code: ForwardErrorCode): void {
        connection.cancel_join?.(new RelayError(code, reason));
        const binding = connection.binding;
        const scope = binding?.session.scope ?? connection.joining_session?.scope;
        if (connection.socket.connected) connection.socket.emit('forward_session_closed', {
            ...(scope ? {forward_session_id: scope.forward_session_id} : {}), reason, code});
        this.detach(connection.socket);
        connection.socket.disconnect(true);
    }
    disconnect(socket: Socket, reason = 'relay_routes_closed'): void {
        const connection = this.connections.get(socket);
        if (connection) this.drop(connection, reason, 'CANCELLED');
    }
    can_reply(socket: Socket): boolean {
        const connection = this.connections.get(socket);
        return !!connection && this.writable(connection);
    }
    unknown_event(socket: Socket): void {
        const connection = this.connections.get(socket);
        if (!connection) fail('UNAUTHORIZED', 'Socket is disconnected');
        try {this.rate(connection);}
        catch (error) {this.drop(connection, 'unknown_event_rate_exceeded', 'RATE_LIMITED'); throw error;}
    }
    private writable(connection: Connection): boolean {
        const transport = connection.socket.conn as unknown as {writeBuffer: unknown[]};
        if (transport.writeBuffer.length >= this.limits.max_buffered_packets) {
            this.drop(connection, 'resync_required', 'RESYNC_REQUIRED');
            return false;
        }
        return connection.socket.connected;
    }
    private close_grant(session: Session, grant: Grant, reason: string, code: ForwardErrorCode): void {
        if (session.grants.get(grant.id) !== grant) return;
        session.grants.delete(grant.id);
        session.revoked_grants.set(grant.id, this.tombstone_expiry());
        clearTimeout(grant.timer);
        for (const connection of [...this.connections.values()])
            if (connection.joining_grant === grant) this.drop(connection, reason, code);
        if (grant.web) this.drop(grant.web.connection, reason, code);
        grant.dedupe.clear();
    }
    private close_session(session: Session, reason: string, code: ForwardErrorCode): void {
        if (this.sessions.get(session.scope.forward_session_id) !== session) return;
        this.sessions.delete(session.scope.forward_session_id);
        this.revoked_sessions.set(session.scope.forward_session_id, this.tombstone_expiry());
        clearTimeout(session.timer);
        clearTimeout(session.reclaim);
        for (const connection of [...this.connections.values()])
            if (connection.joining_session === session) this.drop(connection, reason, code);
        for (const pending of [...session.pending]) pending.finish(error_ack(new RelayError(code, reason)));
        for (const grant of session.grants.values()) clearTimeout(grant.timer);
        const sockets = [...session.webs, ...(session.node ? [session.node] : [])];
        session.ready = false;
        for (const binding of sockets) this.drop(binding.connection, reason, code);
        session.grants.clear();
        this.report(session, 'closed', reason);
    }
    private reclaim(session: Session, delay: number, reason: string): void {
        clearTimeout(session.reclaim);
        const owned = timer(() => {
            if (session.reclaim === owned && this.sessions.get(session.scope.forward_session_id) === session && !session.webs.size)
                this.close_session(session, reason, 'SESSION_STALE');
        }, delay);
        session.reclaim = owned;
    }
    private status(session: Session, reason: string): void {
        if (this.sessions.get(session.scope.forward_session_id) !== session) return;
        const data = {forward_session_id: session.scope.forward_session_id, node_ready: session.ready, reason};
        for (const web of [...session.webs]) if (this.valid(web)) this.send(web, 'relay_status', data);
        this.report(session, session.ready ? 'ready' : 'unavailable', reason);
    }
    private grant(session: Session, claims: RelayClaims): Grant | undefined {
        if (claims.role !== 'web') return undefined;
        const grant = session.grants.get(claims.grant_id!);
        if (!grant || grant.user_id !== claims.user_id) fail('FORBIDDEN', 'Web grant is not authorized');
        if (grant.expires_at <= Date.now()) {
            this.close_grant(session, grant, 'grant_expired', 'TOKEN_EXPIRED');
            fail('TOKEN_EXPIRED', 'Web grant expired');
        }
        if (claims.permissions.some(permission => !grant.permissions.has(permission))) fail('FORBIDDEN', 'Token exceeds grant permissions');
        return grant;
    }
    private valid(binding: Binding): boolean {
        const {connection, session, grant} = binding;
        if (this.connections.get(connection.socket) !== connection || connection.binding !== binding || !connection.socket.connected)
            return false;
        if (this.sessions.get(session.scope.forward_session_id) !== session) return false;
        if (binding.expires_at <= Date.now() || session.lease_expires_at <= Date.now()
            || (grant && (session.grants.get(grant.id) !== grant || grant.expires_at <= Date.now()))) {
            this.drop(connection, 'authorization_expired', 'TOKEN_EXPIRED');
            return false;
        }
        return binding.claims.role === 'node' ? session.node === binding : grant?.web === binding;
    }
    private binding(socket: Socket, role?: RelayClaims['role']): Binding {
        const connection = this.connections.get(socket);
        if (!connection || connection.busy || !connection.binding || !this.valid(connection.binding))
            fail('UNAUTHORIZED', 'Socket is not bound to an active session');
        if (role && connection.binding.claims.role !== role) fail('FORBIDDEN', 'Socket role cannot perform this operation');
        return connection.binding;
    }
    private ready_data(binding: Binding): JoinRoomReady {
        return {...binding.session.scope, role: binding.claims.role, node_ready: binding.session.ready, expires_at: binding.expires_at};
    }
    private async join_room(connection: Connection, room: string): Promise<void> {
        let cancelled = false;
        let deadline: Timer | undefined;
        const cancellation = new Promise<never>((_resolve, reject) => {
            connection.cancel_join = error => {cancelled = true; reject(error);};
            deadline = timer(() => connection.cancel_join?.(new RelayError('TIMEOUT', 'Room join timed out', true)), this.limits.auth_timeout_ms);
        });
        const joining = Promise.resolve().then(() => connection.socket.join(room)).then(() => {
            // An async adapter can complete after disconnect/cancellation and add the
            // socket back to its room. Clean that late membership without retaining it.
            if (cancelled) void Promise.resolve(connection.socket.leave(room)).catch(error => this.report_error(error));
        });
        try {await Promise.race([joining, cancellation]);}
        finally {clearTimeout(deadline); connection.cancel_join = undefined;}
    }
    async join(socket: Socket, command: unknown, reauth = false): Promise<JoinRoomReady> {
        this.control_required();
        const connection = this.connections.get(socket);
        if (!connection || !socket.connected) fail('UNAUTHORIZED', 'Socket is disconnected');
        this.rate(connection);
        if (connection.busy) fail('INVALID_REQUEST', 'Another authentication operation is pending');
        if (!is_record(command) || Object.keys(command).length !== 1 || typeof command.token !== 'string')
            fail('INVALID_REQUEST', 'Expected a token');
        connection.busy = true;
        let joined = false;
        let room: string | undefined;
        try {
            const claims = verify_token(this.options.key, command.token, this.options.forward_id, this.limits);
            const session = this.session(claims.forward_session_id);
            if (!same_scope(session.scope, claims)) fail('SESSION_STALE', 'Token scope differs from the registered session');
            const grant = this.grant(session, claims);
            connection.joining_session = session;
            connection.joining_grant = grant;
            const previous = connection.binding;
            if (previous && (!same_identity(previous.claims, claims) || !this.valid(previous)))
                fail('FORBIDDEN', 'Socket identity cannot change');
            if (previous && reauth && (previous.claims.permissions.length !== claims.permissions.length
                || claims.permissions.some(permission => !previous.claims.permissions.includes(permission))))
                fail('FORBIDDEN', 'Reauthorization cannot change permissions');
            if (reauth && !previous) fail('UNAUTHORIZED', 'Join before renewing authorization');
            if (previous && !reauth) return this.ready_data(previous);
            const expires_at = Math.min(claims.expires_at, session.lease_expires_at, grant?.expires_at ?? Infinity);
            if (expires_at <= Date.now()) fail('TOKEN_EXPIRED', 'Authorization expired');
            if (previous && expires_at < previous.expires_at) fail('INVALID_REQUEST', 'Reauthorization can only extend the lease');
            room = `relay:${session.scope.forward_id}:${session.scope.node_id}:${session.scope.bot_id}:${session.scope.node_session_id}:${session.scope.forward_session_id}:${claims.role}`;
            if (!previous) {await this.join_room(connection, room); joined = true;}
            this.control_required();
            this.active(session);
            if (!socket.connected || this.connections.get(socket) !== connection) fail('CANCELLED', 'Connection closed while joining');
            if (this.grant(session, claims) !== grant) fail('FORBIDDEN', 'Grant was replaced while joining');
            if (expires_at <= Date.now()) fail('TOKEN_EXPIRED', 'Authorization expired while joining');
            if (previous) {
                previous.claims = claims;
                previous.expires_at = expires_at;
                this.expiry(previous);
                return this.ready_data(previous);
            }
            const binding: Binding = {connection, session, claims, grant, expires_at, generation: session.generation,
                subscriptions: new Set(), pending: new Set()};
            connection.binding = binding;
            clearTimeout(connection.auth_timer);
            if (claims.role === 'node') {
                const old = session.node;
                binding.generation = ++session.generation;
                session.node = binding;
                session.ready = false;
                session.stream = undefined;
                session.seq = undefined;
                session.retired = undefined;
                for (const pending of [...session.pending]) pending.finish(error_ack(new RelayError('NODE_OFFLINE', 'Node connection replaced', true)));
                if (old) this.drop(old.connection, 'node_replaced', 'SESSION_STALE');
                this.status(session, 'node_joined');
            } else {
                const old = grant!.web;
                for (const [id, entry] of grant!.dedupe) if (entry.local) grant!.dedupe.delete(id);
                grant!.web = binding;
                session.webs.add(binding);
                session.had_web = true;
                clearTimeout(session.reclaim);
                session.reclaim = undefined;
                if (old) this.drop(old.connection, 'grant_connection_replaced', 'SESSION_STALE');
            }
            this.expiry(binding);
            return this.ready_data(binding);
        } catch (error) {
            if (joined && room && !connection.binding)
                void Promise.resolve(socket.leave(room)).catch(error => this.report_error(error));
            throw error;
        } finally {connection.busy = false; connection.joining_session = undefined; connection.joining_grant = undefined;}
    }
    private expiry(binding: Binding): void {
        clearTimeout(binding.timer);
        binding.timer = timer(() => {
            if (binding.connection.binding === binding) this.drop(binding.connection, 'authorization_expired', 'TOKEN_EXPIRED');
        }, binding.expires_at - Date.now());
    }
    node_ready(socket: Socket, command: unknown): {node_ready: true} {
        const binding = this.binding(socket, 'node');
        const session = binding.session;
        if (!is_record(command) || Object.keys(command).length !== 1 || command.forward_session_id !== session.scope.forward_session_id)
            fail('INVALID_REQUEST', 'Invalid Node readiness message');
        if (!session.ready) {
            session.ready = true;
            this.status(session, 'node_ready');
            if (!session.first_ready) {
                session.first_ready = true;
                if (!session.had_web && !session.webs.size) this.reclaim(session, this.limits.unused_timeout_ms, 'unused_session');
            }
        }
        return {node_ready: true};
    }
    private rate(connection: Connection): void {
        const now = Date.now();
        if (now - connection.request_window >= 1000) {connection.request_window = now; connection.requests = 0;}
        if (++connection.requests > this.limits.requests_per_second) fail('RATE_LIMITED', 'Request rate exceeded', true);
    }
    async request(socket: Socket, command: unknown): Promise<Result> {
        const binding = this.binding(socket, 'web');
        this.rate(binding.connection);
        validate_request(command, this.limits);
        const request = command;
        const permission = request.method === 'get_log_history' ? 'bot.logs'
            : request.method === 'tab_complete' ? 'bot.chat'
            : request.method === 'get_context_info' || request.method === 'world.snapshot' || request.method === 'world.chunk' || request.method === 'world.unwatch' ? 'bot.snapshot'
            : request.method === 'get_data_file_list' || request.method === 'get_file_data' ? 'bot.files'
            : request.method === 'workflow.start' || request.method === 'workflow.stop' || request.method === 'workflow.control' ? 'workflow.control'
            : request.method === 'workflow.snapshot' || request.method === 'workflow.catalog'
                || request.method === 'workflow.command' || request.method === 'workflow.runtime' ? 'bot.snapshot' : request.method;
        if (!binding.claims.permissions.includes(permission) || !binding.grant!.permissions.has(permission))
            fail('FORBIDDEN', 'Method is not authorized');
        const workflow_function = request.method === 'execute_function'
            && (request.payload as {function: string}).function.split('/')[0] === 'workflow';
        if (workflow_function && (!binding.claims.permissions.includes('workflow.control') || !binding.grant!.permissions.has('workflow.control')))
            fail('FORBIDDEN', 'Workflow functions require workflow control permission');
        const non_replayable = request.method === 'bot.chat' || request.method === 'execute_function'
            || request.method === 'workflow.start' || request.method === 'workflow.stop' || request.method === 'workflow.control';
        if (non_replayable || request.method === 'tab_complete') this.control_required();
        const now = Date.now();
        const dedupe = binding.grant!.dedupe;
        const file_request = request.method === 'get_data_file_list' || request.method === 'get_file_data';
        const fresh_read = file_request || request.method.startsWith('world.') || request.method.startsWith('workflow.') && !non_replayable;
        for (const [id, entry] of dedupe) if (entry.expires_at <= now && (entry.completed || entry.result)) dedupe.delete(id);
        const fingerprint = createHash('sha256').update(canonical({method: request.method, payload: request.payload})).digest('hex');
        const existing = dedupe.get(request.request_id);
        if (existing) {
            if (existing.fingerprint !== fingerprint) fail('INVALID_REQUEST', 'Request ID was already used with a different payload');
            if (!existing.completed && !existing.result) fail('RATE_LIMITED', 'Request is already pending', true);
            if (existing.non_replayable) fail('CANCELLED', 'Mutating requests cannot be replayed');
            // Runtime reads must reflect the current Node and its permissions.
            if (request.method !== 'get_context_info' && !fresh_read) return existing.result!;
        }
        if (!existing && dedupe.size >= this.limits.max_dedupe_entries) {
            const oldest = [...dedupe].find(([, entry]) => entry.completed || entry.result);
            if (oldest) dedupe.delete(oldest[0]);
            else fail('RATE_LIMITED', 'Request deduplication capacity reached', true);
        }
        const entry: DedupeEntry = {fingerprint, expires_at: now + this.limits.dedupe_ttl_ms, non_replayable};
        if (request.method === 'bot.subscribe' || request.method === 'bot.unsubscribe') {
            entry.local = true;
            const events = (request.payload as {events: BotEventName[]}).events;
            if (request.method === 'bot.subscribe') binding.subscriptions = new Set(events);
            else for (const event of events) binding.subscriptions.delete(event);
            entry.result = success({events: [...binding.subscriptions]});
            dedupe.set(request.request_id, entry);
            return entry.result;
        }
        const session = binding.session;
        const node = session.node;
        if (!node || !session.ready || !this.valid(node)) fail('NODE_OFFLINE', 'Node relay is not ready', true);
        if (!node.claims.permissions.includes(permission)) fail('FORBIDDEN', 'Node token does not permit this method');
        if (workflow_function && !node.claims.permissions.includes('workflow.control')) fail('FORBIDDEN', 'Node token does not permit workflow functions');
        if (binding.pending.size >= this.limits.max_pending_per_web || this.pending.size >= this.limits.max_pending_total)
            fail('RATE_LIMITED', 'Pending request capacity reached', true);
        dedupe.set(request.request_id, entry);
        return new Promise<Result>(resolve => {
            const id = randomUUID();
            let done = false;
            const pending: Pending = {id, binding, node, session, generation: node.generation, entry, finish: result => {
                if (done) return;
                done = true;
                clearTimeout(pending.timer);
                this.pending.delete(id);
                session.pending.delete(pending);
                binding.pending.delete(pending);
                pending.cancel_transport?.();
                entry.completed = true;
                if (!fresh_read) entry.result = result;
                entry.expires_at = Date.now() + this.limits.dedupe_ttl_ms;
                resolve(result);
            }};
            this.pending.set(id, pending);
            session.pending.add(pending);
            binding.pending.add(pending);
            pending.timer = timer(() => pending.finish(error_ack(new RelayError('TIMEOUT', 'Node request timed out', true))), this.limits.rpc_timeout_ms);
            void this.dispatch(pending, request);
        });
    }
    private async dispatch(pending: Pending, request: RelayRequest): Promise<void> {
        const {session, node, binding} = pending;
        try {
            if ((node.connection.socket.conn as unknown as {writeBuffer: unknown[]}).writeBuffer.length >= this.limits.max_buffered_packets) {
                pending.finish(error_ack(new RelayError('RESYNC_REQUIRED', 'Node transport buffer exceeded its limit')));
                this.drop(node.connection, 'resync_required', 'RESYNC_REQUIRED');
                return;
            }
            // Socket.IO has no public ACK cancellation API. Release its ACK wrapper as
            // well as our own timer when a grant, Web, or Node is revoked/replaced.
            const acks = (node.connection.socket as unknown as {acks: Map<number, (...args: unknown[]) => void>}).acks;
            const before = new Set(acks.keys());
            const response_promise = node.connection.socket.timeout(this.limits.rpc_timeout_ms).emitWithAck('relay_request', {
                ...request, correlation_id: pending.id, forward_session_id: session.scope.forward_session_id,
                deadline_at: Date.now() + this.limits.rpc_timeout_ms});
            const ack_id = [...acks.keys()].find(id => !before.has(id));
            pending.cancel_transport = () => {
                if (ack_id === undefined) return;
                const callback = acks.get(ack_id);
                acks.delete(ack_id);
                callback?.(error_ack(new RelayError('CANCELLED', 'Relay request was cancelled')));
            };
            const response = await response_promise;
            if (!this.pending.has(pending.id)) return;
            if (session.node !== node || session.generation !== pending.generation || !this.valid(node))
                fail('NODE_OFFLINE', 'Node connection changed', true);
            if (!this.valid(binding)) fail('CANCELLED', 'Web authorization ended');
            validate_node_ack(request.method, response, session.scope.bot_id, this.limits, request.payload);
            if ((request.method === 'workflow.control' || request.method === 'workflow.command')
                && response.status === 'success' && response.data !== null) {
                const input = request.payload as {command_id: string; action?: string; workflow_id?: string};
                const data = response.data as {command_id: string; action: string; workflow_id: string | null};
                if (data.command_id !== input.command_id || (request.method === 'workflow.control'
                    && (data.action !== input.action || (input.workflow_id !== undefined && data.workflow_id !== input.workflow_id))))
                    fail('INVALID_REQUEST', 'Node workflow command does not match the request');
            }
            if (request.method === 'workflow.snapshot' && response.status === 'success') {
                const {run_id} = request.payload as {run_id?: string};
                if (run_id === undefined ? !Array.isArray(response.data)
                    : response.data !== null && (Array.isArray(response.data) || (response.data as {run_id: string}).run_id !== run_id))
                    fail('INVALID_REQUEST', 'Node workflow snapshot does not match the request');
            }
            if (request.method === 'get_file_data' && response.status === 'success') {
                const input = request.payload as {id: string; offset?: number; length?: number};
                const data = response.data as {id: string; offset: number; next_offset: number};
                if (data.id !== input.id || data.offset !== (input.offset ?? 0)
                    || data.next_offset - data.offset > (input.length ?? 262_144))
                    fail('INVALID_REQUEST', 'Node file response does not match the request');
            }
            if (request.method === 'tab_complete' && response.status === 'success') {
                const {start, length} = response.data as {start: number; length: number};
                if (start + length > (request.payload as {text: string}).text.length)
                    fail('INVALID_REQUEST', 'Node completion range exceeds the requested text');
            }
            if (request.method === 'bot.logs' && response.status === 'success'
                && (response.data as {records: unknown[]}).records.length > (request.payload as {limit: number}).limit)
                fail('INVALID_REQUEST', 'Node logs response exceeds the requested limit');
            pending.finish(response);
        } catch (error) {
            if (!this.pending.has(pending.id)) return;
            pending.finish(error_ack(error instanceof RelayError ? error : new RelayError('TIMEOUT', 'Node request failed or timed out', true)));
        }
    }
    private send(binding: Binding, event: string, value: unknown): void {
        const socket = binding.connection.socket;
        if (!this.writable(binding.connection)) return;
        socket.emit(event, value);
    }
    private event_binding(socket: Socket): Binding {
        const binding = this.binding(socket, 'node');
        const connection = binding.connection;
        const now = Date.now();
        if (!connection.log_window || now - connection.log_window >= 1000) {
            connection.log_window = now;
            connection.logs = 0;
        }
        connection.logs = (connection.logs ?? 0) + 1;
        if (connection.logs > this.limits.events_per_second) {
            this.drop(connection, 'log_rate_exceeded', 'RATE_LIMITED');
            fail('RATE_LIMITED', 'Node log rate exceeded', true);
        }
        const session = binding.session;
        if (!session.ready) fail('NODE_OFFLINE', 'Node has not declared readiness', true);
        return binding;
    }
    log(socket: Socket, command: unknown): {accepted: true} {
        const binding = this.event_binding(socket);
        validate_event(command, this.limits);
        if (command.name !== 'log') fail('INVALID_REQUEST', 'Expected a log event');
        return this.fanout(binding, command, 'bot.logs', 'bot_log');
    }
    event(socket: Socket, command: unknown): {accepted: true} {
        const binding = this.event_binding(socket);
        validate_event(command, this.limits);
        if (command.name !== 'status_update') fail('INVALID_REQUEST', 'Expected a status update event');
        if (!binding.claims.permissions.includes('bot.snapshot')) fail('FORBIDDEN', 'Node token does not permit status updates');
        return this.fanout(binding, command, 'bot.snapshot', 'bot_event');
    }
    info(socket: Socket, command: unknown): {accepted: true} {
        const binding = this.event_binding(socket);
        validate_info_update(command, this.limits);
        if (!binding.claims.permissions.includes('bot.snapshot')) fail('FORBIDDEN', 'Node token does not permit info updates');
        return this.fanout(binding, command, 'bot.snapshot', 'info_update');
    }
    private fanout<K extends keyof RelayEventPayloads>(binding: Binding, command: RelayEventPayloads[K], permission: RelayMethod,
        event: K): {accepted: true} {
        const {session, connection} = binding;
        const label = event === 'bot_log' ? 'Log' : event === 'info_update' ? 'Info' : 'Lifecycle';
        const kind = event === 'bot_log' ? 'log' : event === 'info_update' ? 'info' : 'lifecycle';
        if (command.forward_session_id !== session.scope.forward_session_id || command.bot_id !== session.scope.bot_id)
            fail('FORBIDDEN', `${label} scope does not match the Node connection`);
        if (session.stream !== command.stream_id) {
            session.retired ??= new Set();
            if (session.retired.has(command.stream_id)) fail('RESYNC_REQUIRED', `Retired ${kind} stream cannot resume`);
            if (session.stream) {
                if (session.retired.size >= 32) {
                    this.drop(connection, 'resync_required', 'RESYNC_REQUIRED');
                    fail('RESYNC_REQUIRED', `Too many ${kind} stream replacements`);
                }
                session.retired.add(session.stream);
                this.status(session, 'resync_required');
            }
            session.stream = command.stream_id;
            session.seq = -1;
        }
        if (command.seq <= (session.seq ?? -1)) fail('INVALID_REQUEST', `${label} sequence must increase`);
        session.seq = command.seq;
        for (const web of [...session.webs]) {
            if (this.valid(web) && web.claims.permissions.includes(permission) && web.grant!.permissions.has(permission))
                this.send(web, event, command);
        }
        return {accepted: true};
    }

    close(): void {
        if (this.closed) return;
        this.closed = true;
        this.control = false;
        for (const session of [...this.sessions.values()]) this.close_session(session, 'relay_closed', 'RELAY_UNAVAILABLE');
        for (const connection of [...this.connections.values()]) this.drop(connection, 'relay_closed', 'RELAY_UNAVAILABLE');
        this.revoked_sessions.clear();
    }
}

export function create_session_store(options: SessionStoreOptions): RelaySessionStore {return new RelaySessionStore(options);}