/**
 * @license
 * Copyright 2022-2026 Matter.js Authors
 * SPDX-License-Identifier: Apache-2.0
 */

import { NodeActivity } from "#behavior/context/NodeActivity.js";
import { RemoteActorContext } from "#behavior/context/server/RemoteActorContext.js";
import { AccessControlServer } from "#behaviors/access-control";
import {
    Crypto,
    Diagnostic,
    Duration,
    Entropy,
    InternalError,
    Lifetime,
    Logger,
    MatterError,
    MaybePromise,
    Millis,
    Observable,
    Seconds,
} from "@matter/general";
import { GLOBAL_IDS, Specification } from "@matter/model";
import {
    DataReport,
    DataReportPayloadIterator,
    ExchangeManager,
    GroupSession,
    InteractionRecipient,
    InteractionServerMessenger,
    Invoke,
    InvokeRequest,
    InvokeResponseForSend,
    InvokeResult,
    Mark,
    Message,
    MessageExchange,
    MessageType,
    NodeSession,
    PeerAddress,
    ProtocolHandler,
    ReadRequest,
    Session,
    SessionManager,
    SessionType,
    Subject,
    SubscribeRequest,
    Subscription,
    TimedRequest,
    WriteRequest,
    WriteResponse,
    WriteResult,
} from "@matter/protocol";
import {
    AttributeData,
    AttributePath,
    DEFAULT_MAX_PATHS_PER_INVOKE,
    DelayReportData,
    EndpointNumber,
    EventPath,
    GroupId,
    INTERACTION_PROTOCOL_ID,
    InvokeResponseData,
    ReceivedStatusResponseError,
    Status,
    StatusResponseError,
    TlvAny,
    TlvClusterPath,
    TlvInvokeResponseData,
    TlvInvokeResponseForSend,
    TlvSubscribeResponse,
    TypeFromSchema,
} from "@matter/types";
import { Groupcast } from "@matter/types/clusters/groupcast";
import { ServerNode } from "../ServerNode.js";
import { OnlineServerInteraction } from "./OnlineServerInteraction.js";
import { ServerSubscription, ServerSubscriptionConfig, ServerSubscriptionContext } from "./ServerSubscription.js";

const logger = Logger.get("InteractionServer");

function isConcreteWritePath(path: AttributePath): path is WriteResult.ConcreteAttributePath {
    return path.endpointId !== undefined && path.clusterId !== undefined && path.attributeId !== undefined;
}

function writePathKey({ endpointId, clusterId, attributeId }: WriteResult.ConcreteAttributePath) {
    return `${endpointId}/${clusterId}/${attributeId}`;
}

// Hard acceptance ceiling for paths in a single read/subscribe interaction, set to the upper bound of the
// CapabilityMinimaStruct field constraints in the data model.  Separate from the advertised CapabilityMinima defaults:
// we advertise a guaranteed floor and accept up to this ceiling, blocking only beyond it.
const MAX_READ_PATHS = 10_000;
const MAX_SUBSCRIBE_PATHS = 10_000;

// A controller decides for itself when its sessions are gone; a report we cannot push does not prove they are
const SUBSCRIPTION_EXCHANGE_OPTIONS: MessageExchange.Options = { suppressPeerLoss: true };

export interface PeerSubscription {
    subscriptionId: number;
    peerAddress: PeerAddress;
    minIntervalFloor: Duration;
    maxIntervalCeiling: Duration;
    attributeRequests?: AttributePath[];
    eventRequests?: EventPath[];
    isFabricFiltered: boolean;
    maxInterval: Duration;
    sendInterval: Duration;
}

/** Cumulative counters tracked by the interaction server since startup. */
export interface InteractionCounters {
    totalSubscriptionsEstablished: number;
    totalInteractionModelMessagesSent: number;
    totalInteractionModelMessagesReceived: number;
}

function validateReadAttributesPath(path: AttributePath) {
    const { clusterId, attributeId } = path;
    if (clusterId === undefined && attributeId !== undefined) {
        if (!GLOBAL_IDS.has(attributeId)) {
            throw new StatusResponseError(
                `Illegal read request for wildcard cluster and non global attribute ${attributeId}`,
                Status.InvalidAction,
            );
        }
    }
}

function validateReadEventPath(path: EventPath) {
    const { clusterId, eventId } = path;
    if (clusterId === undefined && eventId !== undefined) {
        throw new StatusResponseError("Illegal read request with wildcard cluster ID", Status.InvalidAction);
    }
}

/** Per Matter §8.4.3.2, Read and Subscribe share one valid-path algorithm. */
function validateReadPaths(attributeRequests?: AttributePath[], eventRequests?: EventPath[]) {
    attributeRequests?.forEach(path => validateReadAttributesPath(path));
    eventRequests?.forEach(path => validateReadEventPath(path));
}

function clusterPathToId({ nodeId, endpointId, clusterId }: TypeFromSchema<typeof TlvClusterPath>) {
    return `${nodeId}/${endpointId}/${clusterId}`;
}

/**
 * Interfaces {@link InteractionServer} with other components.
 */
export interface InteractionContext {
    readonly sessions: SessionManager;
    readonly exchangeManager: ExchangeManager;
}

/**
 * Translates interactions from the Matter protocol to matter.js APIs.
 */
export class InteractionServer implements ProtocolHandler, InteractionRecipient {
    readonly #lifetime: Lifetime;
    readonly id = INTERACTION_PROTOCOL_ID;
    readonly requiresSecureSession = true;
    #context: InteractionContext;
    #nextSubscriptionId: number;
    #isClosing = false;
    #clientHandler?: ProtocolHandler;
    readonly #subscriptionConfig: ServerSubscriptionConfig;
    readonly #maxPathsPerInvoke;
    readonly #subscriptionEstablishmentStarted = Observable<[peerAddress: PeerAddress]>();
    #node: ServerNode;
    #activity: NodeActivity;
    #newActivityBlocked = false;
    #aclServer?: AccessControlServer;
    #serverInteraction: OnlineServerInteraction;

    /**
     * The exchange of each chunked write interaction, by the attributes it is writing.  Between chunks only the last
     * attribute stays held.  Another exchange's write to a held attribute is answered BUSY or skipped, as the CHIP SDK
     * does, so it cannot land between the chunks of a list write.
     */
    readonly #writingExchanges = new Map<string, MessageExchange>();
    #counters: InteractionCounters = {
        totalSubscriptionsEstablished: 0,
        totalInteractionModelMessagesSent: 0,
        totalInteractionModelMessagesReceived: 0,
    };

    constructor(node: ServerNode, sessions: SessionManager) {
        this.#lifetime = node.construction.join("interaction server");

        this.#nextSubscriptionId = node.env.get(Crypto).randomUint32;

        this.#context = {
            sessions,
            exchangeManager: node.env.get(ExchangeManager),
        };

        this.#subscriptionConfig = ServerSubscriptionConfig.of(node.state.network.subscriptionOptions);
        this.#maxPathsPerInvoke = node.state.basicInformation.maxPathsPerInvoke ?? DEFAULT_MAX_PATHS_PER_INVOKE;

        this.#activity = node.env.get(NodeActivity);

        this.#node = node;

        // ServerInteraction is the "new way" and will replace most logic here over time and especially
        // the InteractionEndpointStructure, which is currently a duplication of the node protocol
        this.#serverInteraction = new OnlineServerInteraction(node.protocol, {
            isHeldByOtherWrite: (path, { exchange }) => this.#isHeldByOther(path, exchange),
        });
    }

    #isHeldByOther(path: WriteResult.ConcreteAttributePath, exchange: MessageExchange) {
        const writing = this.#writingExchanges.get(writePathKey(path));
        return writing !== undefined && writing !== exchange;
    }

    async [Symbol.asyncDispose]() {
        await this.close();
    }

    blockNewActivity() {
        this.#newActivityBlocked = true;
    }

    protected get isClosing() {
        return this.#isClosing;
    }

    get maxPathsPerInvoke() {
        return this.#maxPathsPerInvoke;
    }

    get subscriptionEstablishmentStarted() {
        return this.#subscriptionEstablishmentStarted;
    }

    get counters(): InteractionCounters {
        return this.#counters;
    }

    async onNewExchange(exchange: MessageExchange, message: Message) {
        // When closing, ignore anything newly incoming
        if (this.#newActivityBlocked || this.isClosing) {
            return;
        }

        // An incoming data report as the first message is not a valid server operation.  We instead delegate to a
        // client implementation if available
        if (message.payloadHeader.messageType === MessageType.ReportData && this.clientHandler) {
            return await this.clientHandler.onNewExchange(exchange, message);
        }

        // Count the initiating message (received before the exchange notifier could be attached) then track all
        // further sends/receipts on this exchange (chunked DataReports, StatusResponse acks, etc.)
        this.#counters.totalInteractionModelMessagesReceived++;
        this.#countExchangeMessages(exchange);

        // Activity tracking.  This provides diagnostic information and prevents the server from shutting down whilst
        // the exchange is active
        using activity = this.#activity.begin(`session#${exchange.session.id.toString(16)}`);
        (exchange as NodeActivity.WithActivity)[NodeActivity.activityKey] = activity;

        // Delegate to InteractionServerMessenger
        try {
            return await new InteractionServerMessenger(exchange).handleRequest(this);
        } finally {
            delete (exchange as NodeActivity.WithActivity)[NodeActivity.activityKey];
        }
    }

    /**
     * Attach per-message counters to an exchange so DeviceLoadStatus reflects every IM message sent and received.
     * Excludes MRP retransmissions (only the initial transmission of each message counts) and duplicate receipts.
     */
    #countExchangeMessages(exchange: MessageExchange) {
        exchange.onSend = (_, retransmission) => {
            if (retransmission === 0) {
                this.#counters.totalInteractionModelMessagesSent++;
            }
        };
        exchange.onReceive = (_, duplicate) => {
            if (!duplicate) {
                this.#counters.totalInteractionModelMessagesReceived++;
            }
        };
    }

    get aclServer() {
        if (this.#aclServer !== undefined) {
            return this.#aclServer;
        }
        const aclServer = this.#node.act(agent => agent.get(AccessControlServer));
        if (MaybePromise.is(aclServer)) {
            throw new InternalError("AccessControlServer should already be initialized.");
        }
        return (this.#aclServer = aclServer);
    }

    get clientHandler(): ProtocolHandler | undefined {
        return this.#clientHandler;
    }

    set clientHandler(clientHandler: ProtocolHandler) {
        this.#clientHandler = clientHandler;
    }

    #prepareOnlineContext(
        exchange: MessageExchange,
        message?: Message,
        fabricFiltered?: boolean,
        timed = false,
    ): RemoteActorContext.Options {
        return {
            activity: (exchange as NodeActivity.WithActivity)[NodeActivity.activityKey],
            fabricFiltered,
            timed,
            message,
            exchange,
            node: this.#node,
        };
    }

    /**
     * Returns an iterator that yields the data reports and events data for the given read request.
     */
    async *#executeReadInteraction(readRequest: ReadRequest, exchange: MessageExchange, message: Message) {
        const readContext = this.#prepareOnlineContext(exchange, message, readRequest.isFabricFiltered);

        for await (const chunk of this.#serverInteraction.read(readRequest, readContext)) {
            for await (const report of chunk) {
                yield InteractionServerMessenger.convertServerInteractionReport(report);
            }
        }
    }

    async handleReadRequest(
        exchange: MessageExchange,
        readRequest: ReadRequest,
        message: Message,
    ): Promise<{ dataReport: DataReport; payload?: DataReportPayloadIterator }> {
        const { attributeRequests, eventRequests, isFabricFiltered, dataVersionFilters, eventFilters } = readRequest;

        logger.debug(() => [
            "Read",
            Mark.INBOUND,
            exchange.via,
            Diagnostic.asFlags({ fabricFiltered: isFabricFiltered }),
            Diagnostic.dict({
                attributes: `${
                    attributeRequests?.map(path => this.#node.protocol.inspectPath(path)).join(", ") ?? "none"
                }${dataVersionFilters?.length ? ` with ${dataVersionFilters?.length} filters` : ""}`,
                events: `${
                    eventRequests?.map(path => this.#node.protocol.inspectPath(path)).join(", ") ?? "none"
                }${eventFilters?.length ? `, ${eventFilters?.length} filters` : ""}`,
            }),
        ]);

        if (attributeRequests === undefined && eventRequests === undefined) {
            return {
                dataReport: {
                    interactionModelRevision: Specification.INTERACTION_MODEL_REVISION,
                    suppressResponse: true,
                },
            };
        }

        if (message.packetHeader.sessionType !== SessionType.Unicast) {
            throw new StatusResponseError(
                "Reads are only allowed on unicast sessions", // Means "No groups"
                Status.InvalidAction,
            );
        }

        validateReadPaths(attributeRequests, eventRequests);

        const readPathsCount = (attributeRequests?.length ?? 0) + (eventRequests?.length ?? 0);
        if (readPathsCount > MAX_READ_PATHS) {
            throw new StatusResponseError(
                `Read request with ${readPathsCount} paths exceeds maximum of ${MAX_READ_PATHS}`,
                Status.PathsExhausted,
            );
        }

        return {
            dataReport: {
                interactionModelRevision: Specification.INTERACTION_MODEL_REVISION,
                suppressResponse: true,
            },
            payload: this.#executeReadInteraction(readRequest, exchange, message),
        };
    }

    async handleWriteRequest(
        exchange: MessageExchange,
        writeRequest: WriteRequest,
        messenger: InteractionServerMessenger,
        message: Message,
    ): Promise<void> {
        let { suppressResponse, writeRequests, moreChunkedMessages } = writeRequest;
        const { timedRequest } = writeRequest;
        const sessionType = message.packetHeader.sessionType;

        logger.info(() => [
            "Write",
            Mark.INBOUND,
            exchange.via,
            Diagnostic.asFlags({ suppressResponse, moreChunkedMessages }),
            Diagnostic.weak(writeRequests.map(req => this.#node.protocol.inspectPath(req.path)).join(", ")),
        ]);

        if (moreChunkedMessages && suppressResponse) {
            throw new StatusResponseError(
                "MoreChunkedMessages and SuppressResponse cannot be used together in write messages",
                Status.InvalidAction,
            );
        }

        const receivedWithinTimedInteraction = exchange.hasActiveTimedInteraction();

        if (receivedWithinTimedInteraction && moreChunkedMessages) {
            throw new StatusResponseError(
                "Write Request action that is part of a Timed Write Interaction SHALL NOT be chunked.",
                Status.InvalidAction,
            );
        }

        if (exchange.hasExpiredTimedInteraction()) {
            exchange.clearTimedInteraction();
            throw new StatusResponseError(`Timed request window expired. Decline write request.`, Status.Timeout);
        }

        if (timedRequest !== exchange.hasTimedInteraction()) {
            throw new StatusResponseError(
                `timedRequest flag of write interaction (${timedRequest}) mismatch with expected timed interaction (${receivedWithinTimedInteraction}).`,
                Status.TimedRequestMismatch,
            );
        }

        if (receivedWithinTimedInteraction) {
            logger.debug("Write request for timed interaction on", exchange.channel.name);
            exchange.clearTimedInteraction();
            if (sessionType !== SessionType.Unicast) {
                throw new StatusResponseError(
                    "Write requests are only allowed on unicast sessions when a timed interaction is running.",
                    Status.InvalidAction,
                );
            }
        }

        if (sessionType === SessionType.Group && !suppressResponse) {
            throw new StatusResponseError(
                "Write requests are only allowed as group casts when suppressResponse=true.",
                Status.InvalidAction,
            );
        }

        // Track the previous processed attribute path for list operations across chunks.
        // A list ADD (listIndex === null) is only valid if the previous write was to the same attribute.
        let previousProcessedAttributePath: WriteResult.ConcreteAttributePath | undefined;

        const heldPaths = new Set<string>();
        const releaseHeldPaths = (keep?: string) => {
            for (const key of heldPaths) {
                if (key !== keep) {
                    this.#writingExchanges.delete(key);
                    heldPaths.delete(key);
                }
            }
        };
        using _held = { [Symbol.dispose]: () => releaseHeldPaths() };

        // Process chunks until moreChunkedMessages is false
        while (true) {
            const allResponses = new Array<WriteResult.AttributeStatus>();

            // Separate write requests into batches based on list validity
            // A list ADD without a prior REPLACE_ALL to the same attribute gets a BUSY response
            let currentBatch = new Array<AttributeData>();

            const processBatch = async () => {
                if (currentBatch.length === 0) {
                    return;
                }

                const context = this.#prepareOnlineContext(
                    exchange,
                    message,
                    true, // always fabric filtered
                    receivedWithinTimedInteraction,
                );

                // Send batch to OnlineServerInteraction
                const batchRequest = { ...writeRequest, writeRequests: currentBatch, suppressResponse: false };
                const batchResults = await this.#serverInteraction.write(batchRequest, context);
                if (batchResults) {
                    allResponses.push(...batchResults);
                }

                currentBatch = [];
            };

            for (const request of writeRequests) {
                const { path } = request;
                const listIndex = path.listIndex;

                if (isConcreteWritePath(path) && this.#isHeldByOther(path, exchange)) {
                    await processBatch();
                    allResponses.push({ kind: "attr-status", path, status: Status.Busy });
                    continue;
                }

                if (listIndex === null) {
                    // This is a list ADD - check if a previous path matches
                    if (
                        previousProcessedAttributePath?.endpointId !== path.endpointId ||
                        previousProcessedAttributePath?.clusterId !== path.clusterId ||
                        previousProcessedAttributePath?.attributeId !== path.attributeId
                    ) {
                        // Invalid ADD - process any pending batch first
                        await processBatch();

                        // According to Specification, ADDs are only allowed with a REPLACE before them
                        // Chip SDK returns "BUSY" in cases where this rule is not followed, so we do too
                        allResponses.push({
                            kind: "attr-status",
                            path: path as WriteResult.ConcreteAttributePath,
                            status: Status.Busy,
                        });

                        // Don't update previousProcessedAttributePath for BUSY responses
                        continue;
                    }
                }

                // Valid write - add to batch and update tracking
                currentBatch.push(request);
                if (isConcreteWritePath(path)) {
                    previousProcessedAttributePath = path;
                    // Taken before any await, so no other exchange passes the BUSY check for this path meanwhile
                    if (moreChunkedMessages || heldPaths.size) {
                        const key = writePathKey(path);
                        heldPaths.add(key);
                        this.#writingExchanges.set(key, exchange);
                    }
                }
            }

            // Process any remaining batch
            await processBatch();

            if (suppressResponse) {
                // No response to send, we are done
                break;
            }

            if (moreChunkedMessages) {
                releaseHeldPaths(
                    previousProcessedAttributePath === undefined
                        ? undefined
                        : writePathKey(previousProcessedAttributePath),
                );
            }

            // Send WriteResponse for this chunk
            const chunkResponse: WriteResponse = {
                writeResponses: allResponses.map(({ path, status, clusterStatus }) => ({
                    path,
                    status: { status, clusterStatus },
                })),
                interactionModelRevision: Specification.INTERACTION_MODEL_REVISION,
            };

            await messenger.sendWriteResponse(chunkResponse, {
                logContext: moreChunkedMessages ? "WriteResponse-chunk" : undefined,
            });

            if (!moreChunkedMessages) {
                // Was the last message, so we are done
                break;
            }

            // Wait for the next chunk
            const nextChunk = await messenger.readNextWriteRequest();
            const nextRequest = nextChunk.writeRequest;
            ({ writeRequests, moreChunkedMessages, suppressResponse } = nextRequest);

            logger.info(() => [
                "Write",
                Mark.INBOUND,
                exchange.via,
                Diagnostic.asFlags({ suppressResponse, moreChunkedMessages }),
                Diagnostic.weak(writeRequests.map(req => this.#node.protocol.inspectPath(req.path)).join(", ")),
            ]);

            if (suppressResponse) {
                throw new StatusResponseError(
                    "Multiple chunked messages and SuppressResponse cannot be used together in write messages",
                    Status.InvalidAction,
                );
            }
        }
    }

    async handleSubscribeRequest(
        exchange: MessageExchange,
        request: SubscribeRequest,
        messenger: InteractionServerMessenger,
        message: Message,
    ): Promise<void> {
        const {
            minIntervalFloorSeconds,
            maxIntervalCeilingSeconds,
            attributeRequests,
            dataVersionFilters,
            eventRequests,
            eventFilters,
            keepSubscriptions,
            isFabricFiltered,
        } = request;

        logger.info(() => [
            "Subscribe",
            Mark.INBOUND,
            exchange.via,
            Diagnostic.asFlags({ fabricFiltered: isFabricFiltered, keepSubscriptions }),
            Diagnostic.dict({
                attributePaths: attributeRequests?.length,
                eventPaths: eventRequests?.length,
            }),
        ]);

        if (message.packetHeader.sessionType !== SessionType.Unicast) {
            throw new StatusResponseError("Subscriptions are only allowed on unicast sessions", Status.InvalidAction);
        }

        const subscribePathsCount = (attributeRequests?.length ?? 0) + (eventRequests?.length ?? 0);
        if (subscribePathsCount > MAX_SUBSCRIBE_PATHS) {
            throw new StatusResponseError(
                `Subscribe request with ${subscribePathsCount} paths exceeds maximum of ${MAX_SUBSCRIBE_PATHS}`,
                Status.PathsExhausted,
            );
        }

        NodeSession.assert(exchange.session, "Subscriptions are only implemented on secure sessions");
        const session = exchange.session;
        const fabric = session.fabric;

        if (fabric !== undefined && !keepSubscriptions) {
            let clearedCount = 0;
            for (const sess of this.#context.sessions.sessions) {
                // TODO Adjust this filtering when subscriptions move to Peer
                if (!PeerAddress.is(sess.peerAddress, session.peerAddress)) {
                    // Ignore subscriptions from other peers
                    continue;
                }
                for (const subscription of sess.subscriptions) {
                    await subscription.handlePeerCancel();
                    clearedCount++;
                }
            }
            if (clearedCount > 0) {
                logger.debug(
                    `Cleared ${clearedCount} subscriptions for Subscriber node ${session.peerNodeId} because keepSubscriptions=false`,
                );
            }
        }

        if (
            (!Array.isArray(attributeRequests) || attributeRequests.length === 0) &&
            (!Array.isArray(eventRequests) || eventRequests.length === 0)
        ) {
            throw new StatusResponseError("No attributes or events requested", Status.InvalidAction);
        }

        logger.debug(() => [
            "Subscribe request details",
            Mark.INBOUND,
            exchange.via,
            Diagnostic.dict({
                attributes: attributeRequests?.length
                    ? attributeRequests?.map(path => this.#node.protocol.inspectPath(path)).join(", ")
                    : undefined,
                dataVersionFilters: dataVersionFilters?.length
                    ? dataVersionFilters
                          .map(
                              ({ path: { nodeId, endpointId, clusterId }, dataVersion }) =>
                                  `${clusterPathToId({ nodeId, endpointId, clusterId })}=${dataVersion}`,
                          )
                          .join(", ")
                    : undefined,
                events: eventRequests?.length
                    ? eventRequests.map(path => this.#node.protocol.inspectPath(path)).join(", ")
                    : undefined,
                eventFilters: eventFilters?.length
                    ? eventFilters.map(filter => `${filter.nodeId}/${filter.eventMin}`).join(", ")
                    : undefined,
            }),
        ]);

        validateReadPaths(attributeRequests, eventRequests);

        if (minIntervalFloorSeconds < 0) {
            throw new StatusResponseError(
                "minIntervalFloorSeconds should be greater or equal to 0",
                Status.InvalidAction,
            );
        }
        if (maxIntervalCeilingSeconds < 0) {
            throw new StatusResponseError(
                "maxIntervalCeilingSeconds should be greater or equal to 0",
                Status.InvalidAction,
            );
        }
        if (maxIntervalCeilingSeconds < minIntervalFloorSeconds) {
            throw new StatusResponseError(
                "maxIntervalCeilingSeconds should be greater or equal to minIntervalFloorSeconds",
                Status.InvalidAction,
            );
        }

        if (this.#nextSubscriptionId === 0xffffffff) this.#nextSubscriptionId = 0;
        const subscriptionId = this.#nextSubscriptionId++;

        this.#subscriptionEstablishmentStarted.emit(session.peerAddress);
        let subscription: ServerSubscription;
        try {
            subscription = await this.#establishSubscription(
                subscriptionId,
                request,
                messenger,
                session,
                exchange,
                message,
            );
        } catch (error) {
            const sre = StatusResponseError.of(error);
            // A client refusing the priming report is routine and peer-triggered
            logger[sre instanceof ReceivedStatusResponseError ? "debug" : "warn"](
                `Subscription ${Subscription.idStrOf(subscriptionId)} for session ${session.via}: Error while sending initial data reports:`,
                error instanceof MatterError ? error.message : error,
            );
            if (sre && !(sre instanceof ReceivedStatusResponseError)) {
                logger.info(
                    "Status",
                    Diagnostic.strong(`${Status[sre.code]}(${sre.code})`),
                    Mark.OUTBOUND,
                    exchange.via,
                    exchange.diagnostics,
                    "Error:",
                    Diagnostic.errorMessage(sre),
                );
                await messenger.sendStatus(sre.code, {
                    logContext: {
                        for: "I/SubscriptionSeed-Status",
                    },
                });
            }
            await messenger.close();
            return; // Make sure to not bubble up the exception
        }

        const maxInterval = subscription.maxInterval;

        // Then send the subscription response
        await messenger.send(
            MessageType.SubscribeResponse,
            TlvSubscribeResponse.encode({
                subscriptionId,
                maxInterval: Seconds.of(maxInterval),
                interactionModelRevision: Specification.INTERACTION_MODEL_REVISION,
            }),
            {
                logContext: {
                    ...Subscription.diagnosticOf(subscriptionId),
                    maxInterval: Duration.format(maxInterval),
                },
            },
        );

        // When an error occurs while sending the response, the subscription is not yet active and will be cleaned up by GC
        if (subscription.activate()) {
            this.#counters.totalSubscriptionsEstablished++;
        }
    }

    #initiateSubscriptionExchange(addressOrSession: PeerAddress | Session, protocolId: number) {
        const exchange =
            addressOrSession instanceof Session
                ? this.#context.exchangeManager.initiateExchangeForSession(
                      addressOrSession,
                      protocolId,
                      SUBSCRIPTION_EXCHANGE_OPTIONS,
                  )
                : this.#context.exchangeManager.initiateExchange(
                      addressOrSession,
                      protocolId,
                      SUBSCRIPTION_EXCHANGE_OPTIONS,
                  );

        // Count subscription report messages we push and the acks we receive in response
        this.#countExchangeMessages(exchange);

        return exchange;
    }

    async #establishSubscription(
        id: number,
        request: SubscribeRequest,
        messenger: InteractionServerMessenger,
        session: NodeSession,
        exchange: MessageExchange,
        message: Message,
    ) {
        const context: ServerSubscriptionContext = {
            session,
            node: this.#node,
            initiateExchange: (addressOrSession, protocolId) =>
                this.#initiateSubscriptionExchange(addressOrSession, protocolId),
        };

        const subscription = new ServerSubscription({
            id,
            context,
            request,
            subscriptionOptions: this.#subscriptionConfig,
        });

        const readContext = this.#prepareOnlineContext(exchange, message, request.isFabricFiltered);
        try {
            // Send the initial data report to prime the subscription with initial data
            await subscription.sendInitialReport(messenger, readContext);
        } catch (error) {
            await subscription.close(); // Cleanup
            throw error;
        }

        logger.notice(
            "Subscribe successful",
            Mark.OUTBOUND,
            exchange.via,
            exchange.diagnostics,
            Diagnostic.dict({
                ...Subscription.diagnosticOf(subscription),
                timing: `${Duration.format(subscription.minIntervalFloor)} - ${Duration.format(subscription.maxIntervalCeiling)} => ${Duration.format(subscription.maxInterval)}`,
                sendInterval: Duration.format(subscription.sendInterval),
            }),
        );

        return subscription;
    }

    async establishFormerSubscription(
        {
            subscriptionId,
            attributeRequests,
            eventRequests,
            isFabricFiltered,
            minIntervalFloor,
            maxIntervalCeiling,
            maxInterval,
            sendInterval,
        }: PeerSubscription,
        session: NodeSession,
    ) {
        const exchange = this.#initiateSubscriptionExchange(session, INTERACTION_PROTOCOL_ID);
        const messenger = new InteractionServerMessenger(exchange);

        let subscription: ServerSubscription | undefined;
        try {
            logger.info(
                `Reestablish subscription`,
                Mark.OUTBOUND,
                exchange.via,
                Diagnostic.dict({
                    ...Subscription.diagnosticOf(subscriptionId),
                    isFabricFiltered,
                    maxInterval: Duration.format(maxInterval),
                    sendInterval: Duration.format(sendInterval),
                }),
            );

            const context: ServerSubscriptionContext = {
                session,
                node: this.#node,
                initiateExchange: (addressOrSession, protocolId) =>
                    this.#initiateSubscriptionExchange(addressOrSession, protocolId),
            };

            subscription = new ServerSubscription({
                id: subscriptionId,
                context,
                request: {
                    attributeRequests,
                    eventRequests,
                    isFabricFiltered,
                    minIntervalFloorSeconds: Seconds.of(minIntervalFloor),
                    maxIntervalCeilingSeconds: Seconds.of(maxIntervalCeiling),
                },
                subscriptionOptions: this.#subscriptionConfig,
                useAsMaxInterval: maxInterval,
                useAsSendInterval: sendInterval,
            });

            const readContext = this.#prepareOnlineContext(exchange, undefined, isFabricFiltered);

            // Send initial data report to prime the subscription with initial data
            await subscription.sendInitialReport(
                messenger,
                readContext,
                true, // Do not send status responses because we simulate that the subscription is still established
            );
            if (subscription.activate()) {
                this.#counters.totalSubscriptionsEstablished++;

                logger.notice(
                    `Subscription successfully reestablished`,
                    Mark.OUTBOUND,
                    exchange.via,
                    exchange.diagnostics,
                    Diagnostic.dict({
                        ...Subscription.diagnosticOf(subscriptionId),
                        timing: `${Duration.format(minIntervalFloor)} - ${Duration.format(maxIntervalCeiling)} => ${Duration.format(subscription.maxInterval)}`,
                        sendInterval: Duration.format(subscription.sendInterval),
                    }),
                );
            }
        } catch (error) {
            await subscription?.close(); // Cleanup
            throw error;
        } finally {
            // Reports use their own exchanges; leaving this one open blocks the session from ever closing gracefully
            try {
                await messenger.close();
            } catch (error) {
                logger.warn("Error closing subscription re-establishment exchange", error);
            }
        }
        return subscription;
    }

    static #formatDelayReportData(delayReportData?: DelayReportData) {
        if (delayReportData === undefined) {
            return undefined;
        }
        const { delayMinMs, delayJitterWindowMs } = delayReportData;
        return `${Duration.format(Millis(delayMinMs))}/${Duration.format(Millis(delayJitterWindowMs))}`;
    }

    /**
     * Hold off the next report of every subscription, on any session, that selects one of the endpoints an invoke
     * dispatches to.  The delay is {@link DelayReportData.delayMinMs} plus a random jitter below
     * {@link DelayReportData.delayJitterWindowMs}.
     */
    #deferReports({ delayMinMs, delayJitterWindowMs }: DelayReportData, endpoints: ReadonlySet<EndpointNumber>) {
        if (endpoints.size === 0) {
            return;
        }

        const jitter = delayJitterWindowMs > 0 ? this.#node.env.get(Entropy).randomUint32 % delayJitterWindowMs : 0;
        const delay = Millis(delayMinMs + jitter);

        for (const session of this.#context.sessions.sessions) {
            for (const subscription of session.subscriptions) {
                if (subscription instanceof ServerSubscription && subscription.selectsAnyEndpoint(endpoints)) {
                    subscription.deferReports(delay);
                }
            }
        }
    }

    async handleInvokeRequest(
        exchange: MessageExchange,
        request: InvokeRequest,
        messenger: InteractionServerMessenger,
        message: Message,
    ): Promise<void> {
        const { invokeRequests, timedRequest, suppressResponse, delayReportData } = request;
        logger.info(() => [
            "Invoke",
            Mark.INBOUND,
            exchange.via,
            Diagnostic.asFlags({ suppressResponse, timedRequest }),
            Diagnostic.dict({
                group: message.packetHeader.destGroupId,
                delayReport: InteractionServer.#formatDelayReportData(delayReportData),
                invokes: invokeRequests
                    .map(({ commandPath: { endpointId, clusterId, commandId } }) =>
                        this.#node.protocol.inspectPath({ endpointId, clusterId, commandId }),
                    )
                    .join(", "),
            }),
        ]);

        const receivedWithinTimedInteraction = exchange.hasActiveTimedInteraction();
        if (exchange.hasExpiredTimedInteraction()) {
            exchange.clearTimedInteraction();
            throw new StatusResponseError(`Timed request window expired. Decline invoke request.`, Status.Timeout);
        }

        if (timedRequest !== exchange.hasTimedInteraction()) {
            throw new StatusResponseError(
                `timedRequest flag of invoke interaction (${timedRequest}) mismatch with expected timed interaction (${receivedWithinTimedInteraction}).`,
                Status.TimedRequestMismatch,
            );
        }

        if (receivedWithinTimedInteraction) {
            logger.debug("Invoke request for timed interaction on", exchange.channel.name);
            exchange.clearTimedInteraction();
            if (message.packetHeader.sessionType !== SessionType.Unicast) {
                throw new StatusResponseError(
                    "Invoke requests are only allowed on unicast sessions when a timed interaction is running.",
                    Status.InvalidAction,
                );
            }
        }

        if (invokeRequests.length > this.#maxPathsPerInvoke) {
            throw new StatusResponseError(
                `Only ${this.#maxPathsPerInvoke} invoke requests are supported in one message. This message contains ${invokeRequests.length}`,
                Status.InvalidAction,
            );
        }

        const context = this.#prepareOnlineContext(exchange, message, undefined, receivedWithinTimedInteraction);

        const isGroupSession = message.packetHeader.sessionType === SessionType.Group;

        const invoke: Invoke =
            delayReportData !== undefined && Specification.isForwardFeatureEnabled("delay-report-data")
                ? { ...request, beforeDispatch: endpoints => this.#deferReports(delayReportData, endpoints) }
                : request;

        // Get the invoke-results from the server interaction
        const results = this.#serverInteraction.invoke(invoke, context);

        // For group sessions: consume iterator, report each dispatched command to SessionManager so
        // Groupcast testing (if enabled for the fabric) can emit the GroupcastTesting event
        if (isGroupSession) {
            const rawGroupId = message.packetHeader.destGroupId;
            const groupId = rawGroupId !== undefined ? GroupId(rawGroupId) : undefined;
            const session = exchange.session;
            const fabric = session.associatedFabric;
            // The destination is left to the listener, which derives it from the message's group id: one session
            // serves every group that shares its key set
            const sourceIp = message.receivedFrom;
            const groupSession = GroupSession.is(session) ? session : undefined;
            // The event's ClusterID/ElementID reflect the request; cmd-response paths carry the response command id
            const requestPath = invokeRequests[0]?.commandPath;
            let emitted = false;
            for await (const chunk of results) {
                for (const data of chunk) {
                    if (data.kind !== "cmd-response" && data.kind !== "cmd-status") {
                        continue;
                    }
                    // Emitted => access allowed (denied paths are dropped, see below); AccessAllowed is the
                    // access-control outcome, not command success (Groupcast spec §11.27.7.6.3).
                    this.#context.sessions.emitGroupMessage({
                        result: Groupcast.GroupcastTestResult.Success,
                        fabric,
                        groupId,
                        sourceIp,
                        endpointId: data.path.endpointId,
                        clusterId: requestPath?.clusterId ?? data.path.clusterId,
                        elementId: requestPath?.commandId ?? data.path.commandId,
                        accessAllowed: true,
                    });
                    emitted = true;
                }
            }
            if (!emitted) {
                const subject = groupSession?.subjectFor(message);
                if (subject !== undefined && Subject.isGroup(subject) && !subject.hasValidMapping) {
                    // The authenticating key is not the one mapped for the group: improperly authenticated per
                    // core§11.27.7.6.3, reported without the unauthenticated group id
                    this.#context.sessions.emitGroupMessage({
                        result: Groupcast.GroupcastTestResult.FailedAuth,
                        fabric,
                        sourceIp,
                    });
                } else {
                    // Wildcard expansion produced no dispatches.  Still emit one event per requested invoke path so
                    // observers see the message arrived: the message was authenticated and processed, so per
                    // core§11.27.7.6.3 the result is Success.  When the group has receiving endpoints the expansion
                    // was filtered by access control (AccessAllowed=false); without endpoints no access evaluation
                    // took place and the field is omitted.
                    const memberEndpoints = groupId !== undefined ? fabric.groups.endpoints.get(groupId) : undefined;
                    const accessAllowed =
                        memberEndpoints !== undefined && memberEndpoints.length > 0 ? false : undefined;
                    for (const { commandPath } of invokeRequests) {
                        this.#context.sessions.emitGroupMessage({
                            result: Groupcast.GroupcastTestResult.Success,
                            fabric,
                            groupId,
                            sourceIp,
                            endpointId: commandPath.endpointId,
                            clusterId: commandPath.clusterId,
                            elementId: commandPath.commandId,
                            accessAllowed,
                        });
                    }
                }
            }
            return;
        }

        // Track accumulated responses for the current message
        const currentChunkResponses = new Array<InvokeResponseData>();
        const emptyInvokeResponse: InvokeResponseForSend = {
            suppressResponse: false, // Deprecated but must be present
            interactionModelRevision: Specification.INTERACTION_MODEL_REVISION,
            invokeResponses: [],
        };
        const emptyInvokeResponseLength = TlvInvokeResponseForSend.encode(emptyInvokeResponse).byteLength;
        let messageSize = emptyInvokeResponseLength;
        let chunkedTransmissionTerminated = false;

        /**
         * Send a chunk when the message size limit would be exceeded.
         */
        const sendChunkIfNeeded = async (invokeResponse: InvokeResponseData): Promise<void> => {
            const encodedInvokeResponse = TlvInvokeResponseData.encodeTlv(invokeResponse);
            const invokeResponseBytes = TlvAny.getEncodedByteLength(encodedInvokeResponse);

            // Check if adding this response would exceed message size
            if (messageSize + invokeResponseBytes > exchange.maxPayloadSize && currentChunkResponses.length > 0) {
                logger.debug(
                    "Invoke (chunk)",
                    Mark.OUTBOUND,
                    exchange.via,
                    Diagnostic.dict({ commands: currentChunkResponses.length }),
                );

                const chunkResponse: InvokeResponseForSend = {
                    ...emptyInvokeResponse,
                    invokeResponses: currentChunkResponses.map(r => TlvInvokeResponseData.encodeTlv(r)),
                };

                if (!(await messenger.sendInvokeResponseChunk(chunkResponse))) {
                    chunkedTransmissionTerminated = true;
                    return;
                }

                // Reset for next chunk
                currentChunkResponses.length = 0;
                messageSize = emptyInvokeResponseLength;
            }

            // Add to the current chunk
            currentChunkResponses.push(invokeResponse);
            messageSize += invokeResponseBytes;
        };

        const toInvokeResponseData = (data: InvokeResult.Data): InvokeResponseData => {
            if (data.kind === "cmd-response") {
                const { path: commandPath, commandRef, data: commandFields } = data;
                return { command: { commandPath, commandFields, commandRef } };
            }
            const { path: commandPath, commandRef, status, clusterStatus } = data;
            return { status: { commandPath, status: { status, clusterStatus }, commandRef } };
        };

        // Per spec §8.8.3.2.1 a SuppressResponse invoke still sends a response if a CommandDataIB is generated. Until
        // one is seen we hold generated statuses back; the first CommandDataIB flushes the held statuses and the rest
        // stream normally. Non-suppressed responses stream from the start.
        let suppressedBuffer: InvokeResponseData[] | undefined = suppressResponse ? [] : undefined;

        for await (const chunk of results) {
            if (chunkedTransmissionTerminated) {
                // Client terminated the chunked series, continue consuming but don't send
                continue;
            }

            for (const data of chunk) {
                const responseData = toInvokeResponseData(data);

                if (suppressedBuffer !== undefined) {
                    if (data.kind !== "cmd-response") {
                        suppressedBuffer.push(responseData);
                        continue;
                    }
                    // First CommandDataIB: commit to sending. Flush the held statuses, then stream the rest.
                    for (const held of suppressedBuffer) {
                        await sendChunkIfNeeded(held);
                    }
                    suppressedBuffer = undefined;
                }

                await sendChunkIfNeeded(responseData);
            }
        }

        // SuppressResponse and no CommandDataIB was generated — send nothing.
        if (suppressedBuffer !== undefined) {
            logger.debug("Invoke (suppressed)", Mark.OUTBOUND, exchange.via, Diagnostic.weak("no response sent"));
            return;
        }

        // Send the final response if not already terminated
        if (!chunkedTransmissionTerminated) {
            if (currentChunkResponses.length > 0) {
                logger.debug(
                    "Invoke (final)",
                    Mark.OUTBOUND,
                    exchange.via,
                    Diagnostic.dict({ commands: currentChunkResponses.length }),
                );
            }

            const finalResponse: InvokeResponseForSend = {
                ...emptyInvokeResponse,
                invokeResponses: currentChunkResponses.map(r => TlvInvokeResponseData.encodeTlv(r)),
            };
            await messenger.sendInvokeResponse(finalResponse);
        }
    }

    handleTimedRequest(exchange: MessageExchange, { timeout }: TimedRequest) {
        const interval = Millis(timeout);

        logger.debug(() => [
            "Timed request",
            Mark.INBOUND,
            exchange.via,
            Diagnostic.dict({
                interval: Duration.format(interval),
            }),
        ]);

        exchange.startTimedInteraction(interval);
    }

    async close() {
        this.#isClosing = true;
        this.#lifetime[Symbol.dispose]();
    }
}