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

import { type ClusterBehavior } from "#behavior/cluster/ClusterBehavior.js";
import { LocalActorContext } from "#behavior/context/server/LocalActorContext.js";
import { CommissioningClient } from "#behavior/system/commissioning/CommissioningClient.js";
import { RemoteDescriptor } from "#behavior/system/commissioning/RemoteDescriptor.js";
import { CommissioningDiscovery } from "#behavior/system/controller/discovery/CommissioningDiscovery.js";
import { ContinuousDiscovery } from "#behavior/system/controller/discovery/ContinuousDiscovery.js";
import { Discovery } from "#behavior/system/controller/discovery/Discovery.js";
import { InstanceDiscovery } from "#behavior/system/controller/discovery/InstanceDiscovery.js";
import { PaseDiscovery } from "#behavior/system/controller/discovery/PaseDiscovery.js";
import { IcdClient } from "#behavior/system/icd/IcdClient.js";
import { NetworkClient } from "#behavior/system/network/NetworkClient.js";
import { BasicInformationClient } from "#behaviors/basic-information";
import { BridgedDeviceBasicInformationClient } from "#behaviors/bridged-device-basic-information";
import { IcdManagementClient } from "#behaviors/icd-management";
import { OperationalCredentialsClient } from "#behaviors/operational-credentials";
import { Endpoint } from "#endpoint/Endpoint.js";
import { EndpointContainer } from "#endpoint/properties/EndpointContainer.js";
import { EndpointType } from "#endpoint/type/EndpointType.js";
import { ClientGroup } from "#node/ClientGroup.js";
import { InteractionServer } from "#node/server/InteractionServer.js";
import { ServerNodeStore } from "#storage/server/ServerNodeStore.js";
import {
    CancelablePromise,
    Diagnostic,
    Duration,
    ImplementationError,
    Lifecycle,
    Logger,
    MatterError,
    MaybePromise,
    Minutes,
    Mutex,
    Observable,
    Seconds,
    Time,
    Timestamp,
    UninitializedDependencyError,
} from "@matter/general";
import {
    ClientSubscriptionHandler,
    ClientSubscriptions,
    CommissioningError,
    FabricManager,
    Peer,
    PeerAddress,
    PeerLeftError,
    SessionManager,
} from "@matter/protocol";
import { FabricIndex } from "@matter/types";
import { ClientNode } from "../ClientNode.js";
import type { ServerNode } from "../ServerNode.js";
import { ClientNodeFactory } from "./ClientNodeFactory.js";
import { ClientStructureEvents } from "./ClientStructureEvents.js";

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

const DEFAULT_TTL = Minutes(15);
const EXPIRATION_INTERVAL = Minutes.one;

/**
 * Manages the set of known remote nodes.
 *
 * Remote nodes are either peers (commissioned into a fabric we share) or commissionable.
 */
export class Peers extends EndpointContainer<ClientNode> {
    #expirationInterval?: CancelablePromise;
    #installedSubscriptionHandler?: ClientSubscriptionHandler;
    #mutex = new Mutex(this);
    #closed = false;
    #commissioning = new Set<ClientNode>();

    constructor(owner: ServerNode) {
        super(owner);

        if (!owner.env.has(ClientNodeFactory)) {
            owner.env.set(ClientNodeFactory, new Factory(this));
        }

        owner.env.applyTo(InteractionServer, this.#configureInteractionServer.bind(this));

        this.added.on(this.#handlePeerAdded.bind(this));
        this.deleted.on(this.#manageExpiration.bind(this));

        this.clusterInstalled(BasicInformationClient).on(this.#instrumentBasicInformation.bind(this));
        this.clusterInstalled(IcdManagementClient).on(this.#installIcdClient.bind(this));
        this.clusterInstalled(BridgedDeviceBasicInformationClient).on(
            this.#instrumentBridgedConfigurationVersion.bind(this),
        );

        const lifecycle = owner.lifecycle;
        lifecycle.online.on(this.#nodeOnline.bind(this));
        lifecycle.offline.on(this.#nodeOffline.bind(this));
    }

    /**
     * Load nodes.  Invoked automatically by owner.
     */
    initialize() {
        const factory = this.owner.env.get(ClientNodeFactory);

        const clientStores = this.owner.env.maybeGet(ServerNodeStore)?.clientStores;
        if (clientStores === undefined) {
            throw new UninitializedDependencyError(
                "Peers",
                "are not available because ServerNode initialization is incomplete",
            );
        }

        // Group nodes have an in-memory only store, so all nodes restored here are ClientNode
        for (const id of clientStores.knownIds) {
            this.add(
                factory.create({
                    id,
                    owner: this.owner,
                }),
            );
        }
    }

    async #nodeOnline() {
        for (const peer of this) {
            if (!peer.lifecycle.isCommissioned || peer.maybeStateOf("network")?.isDisabled) {
                continue;
            }
            try {
                await peer.start();
            } catch (e) {
                MatterError.accept(e);
                logger.warn(`Error starting peer ${peer}:`, e);
            }
        }
        this.#manageExpiration();
    }

    async #nodeOffline() {
        this.#cancelExpiration();
        for (const peer of this) {
            await peer.stop();
        }
    }

    /**
     * Find a specific commissionable node, or, if no discovery options are provided, returns the first discovered node.
     * TODO: Allow to provide an array of options for multiple discoveries (e.g. from a Multi QR code).
     */
    locate(options?: Discovery.Options) {
        return new InstanceDiscovery(this.owner, options);
    }

    /**
     * Employ discovery to find a set of commissionable nodes, the options can be used to limit the discovered devices
     * (e.g. just a specific vendor).
     * TODO: Allow to provide multiple identifiers for multiple discoveries (e.g. from a Multi QR code).
     *
     * If you do not provide a timeout value, will search until canceled, and you need to add a listener to
     * {@link Discovery#discovered} or {@link added} to receive discovered nodes.
     */
    discover(options?: Discovery.Options) {
        return new ContinuousDiscovery(this.owner, options);
    }

    /**
     * Find a specific commissionable node and commission.
     */
    commission(options: CommissioningDiscovery.Options) {
        return new CommissioningDiscovery(this.owner as ServerNode, options);
    }

    /**
     * Find a specific commissionable node and establish a PASE session without commissioning.
     *
     * Useful for split-commissioning scenarios where one controller establishes PASE and another
     * performs the commissioning flow, or for raw PASE channel establishment.
     */
    pase(options: PaseDiscovery.Options) {
        return new PaseDiscovery(this.owner as ServerNode, options);
    }

    /**
     * Find or create a {@link ClientNode} for a device described by {@link descriptor}.
     *
     * If a matching node already exists in the peer collection, returns it after refreshing its addresses and
     * discovery data from the supplied descriptor.  Otherwise creates a new node using the descriptor.
     *
     * After calling {@link forDescriptor}, commission the returned node via {@link ClientNode.commission}.
     */
    async forDescriptor(descriptor: RemoteDescriptor): Promise<ClientNode> {
        const factory = this.owner.env.get(ClientNodeFactory);
        let node = factory.find(descriptor);
        if (node !== undefined) {
            // Refresh addresses and discovery data from the new descriptor
            await node.act(agent => {
                agent.commissioning.descriptor = descriptor;
            });
            return node;
        }
        return factory.create({ commissioning: { descriptor } });
    }

    /**
     * Obtain an {@link Observable} that emits when a specific type of endpoint initializes for a peer.
     *
     * This is useful for initializing general behavior on any peer endpoint of the specified type.
     */
    endpointInstalled<T extends EndpointType>(type: T): Observable<[endpoint: Endpoint<T>]> {
        return this.owner.env.get(ClientStructureEvents).endpointInstalled(type);
    }

    /**
     * Obtain a {@link Observable} that emits when a specific type of cluster initializes for a peer.
     *
     * This is useful for initializing general behavior on any peer endpoint with the specified cluster.
     */
    clusterInstalled<T extends ClusterBehavior.Type>(type: T) {
        return this.owner.env.get(ClientStructureEvents).clusterInstalled(type);
    }

    /**
     * Emits when fixed attributes
     */
    override get(id: number | string | PeerAddress) {
        if (typeof id !== "string" && typeof id !== "number") {
            const address = PeerAddress(id);
            for (const node of this) {
                if (node.behaviors.active.some(({ id }) => id === "commissioning")) {
                    const nodeAddress = node.maybeStateOf("commissioning")?.peerAddress as PeerAddress | undefined;
                    if (nodeAddress !== undefined && PeerAddress.is(nodeAddress, address)) {
                        return node;
                    }
                }
            }
            return undefined;
        }

        return super.get(id);
    }

    override get owner() {
        return super.owner as ServerNode;
    }

    override add(node: ClientNode) {
        node.owner = this.owner;

        super.add(node);
    }

    /**
     * Get or create a client node for the given peer address.
     *
     * This is mainly used to communicate to other known nodes on the fabric without having a formal commissioning
     * process.
     */
    async forAddress(peerAddress: PeerAddress, options: Omit<ClientNode.Options, "owner"> = {}) {
        if (!this.owner.env.get(FabricManager).has(peerAddress)) {
            throw new ImplementationError("Cannot register a peer address for a fabric we do not belong to");
        }

        const existing = this.get(peerAddress);
        if (existing) {
            return existing;
        }

        // Serialize creation through the same mutex the cull/leave paths use. Two concurrent callers for the same
        // address would otherwise both pass the check and both create a ClientNode (the awaits below yield between
        // the check and the add).
        return this.#mutex.produce(async () => {
            let node = this.get(peerAddress);
            if (node) {
                return node;
            }

            if (options.id !== undefined) {
                // We want to initialize a node with a provided id. This could be an injected node, so ensure the
                // ClientNodeStore is constructed. Without id the storage is empty anyway because id is newly assigned
                // TODO: Remove when we remove legacy controller
                const store = this.owner.env.get(ServerNodeStore).clientStores.storeForNode(options.id);
                await store.construction;
            }

            // We do not have that node till now, also not persisted, so create it
            const factory = this.owner.env.get(ClientNodeFactory);
            node = factory.create(options, peerAddress);
            await node.construction;
            this.add(node);

            // Nodes we do not commission are not auto-subscribed but enabled
            // But we add the peer address
            await node.set({
                commissioning: { peerAddress: PeerAddress(peerAddress) },
            });

            return node;
        });
    }

    override async close() {
        this.#closed = true;
        await this.#installedSubscriptionHandler?.close();
        this.#cancelExpiration();
        await this.#mutex;
        await super.close();
    }

    #cancelExpiration() {
        if (this.#expirationInterval) {
            this.#expirationInterval.cancel();
            this.#expirationInterval = undefined;
        }
    }

    #handlePeerAdded() {
        if (this.owner.env.has(InteractionServer)) {
            this.#configureInteractionServer();
        }
        this.#manageExpiration();
    }

    /**
     * If required, installs a listener in the environment's {@link InteractionServer} to handle subscription responses.
     */
    #configureInteractionServer() {
        if (
            this.#closed ||
            this.#installedSubscriptionHandler !== undefined ||
            !this.owner.env.has(InteractionServer)
        ) {
            return;
        }

        const subscriptions = this.owner.env.get(ClientSubscriptions);
        const interactionServer = this.owner.env.get(InteractionServer);

        this.#installedSubscriptionHandler = new ClientSubscriptionHandler(subscriptions);
        interactionServer.clientHandler = this.#installedSubscriptionHandler;
    }

    /**
     * Run a commission attempt on {@link node} while protecting the node from the expired-node cull.
     *
     * Serialization is done through the same {@link #mutex} the cull uses: the busy registration runs as a mutex task,
     * which guarantees no cull is in flight when we register and that any subsequent cull observes the busy flag.
     *
     * Rejects with {@link CommissioningError} if the node is already mid-commission (parallel attempts on the same
     * {@link ClientNode} would race on device-side state) or if a cull queued ahead of us already destroyed/crashed
     * the node.  Either way the attempt fails fast instead of crashing later when the closed backing is accessed.
     */
    async runCommissioning<T>(node: ClientNode, fn: () => MaybePromise<T>): Promise<T> {
        await this.#mutex.produce(async () => {
            const status = node.construction.status;
            if (
                status === Lifecycle.Status.Destroying ||
                status === Lifecycle.Status.Destroyed ||
                status === Lifecycle.Status.Crashed
            ) {
                throw new CommissioningError(`Cannot commission ${node.toString()} because the node is ${status}`);
            }
            if (this.#commissioning.has(node)) {
                throw new CommissioningError(
                    `Cannot commission ${node.toString()} because a commission attempt is already in progress`,
                );
            }
            this.#commissioning.add(node);
        });
        try {
            return await fn();
        } finally {
            this.#commissioning.delete(node);
        }
    }

    /**
     * Enables or disables the expiration timer that culls expired uncommissioned nodes.
     */
    #manageExpiration() {
        if (this.#closed) {
            return;
        }

        if (this.#expirationInterval) {
            if (!this.size) {
                this.#cancelExpiration();
            }
            return;
        }

        if (!this.size) {
            return;
        }

        this.#expirationInterval = Time.sleep("client node expiration", EXPIRATION_INTERVAL).then(
            this.#onExpirationIntervalElapsed.bind(this),
        );
    }

    #onExpirationIntervalElapsed() {
        if (this.#closed) {
            return;
        }
        this.#mutex.run(() =>
            this.#cullExpiredNodesAndAddresses()
                .catch(error => {
                    logger.warn("Error culling expired nodes", error);
                })
                .finally(() => {
                    this.#manageExpiration();
                }),
        );
    }

    async #cullExpiredNodesAndAddresses() {
        try {
            const now = Time.nowMs;

            for (const node of this) {
                if (!node.lifecycle.isReady) {
                    continue;
                }
                if (this.#commissioning.has(node)) {
                    continue;
                }
                const state = node.maybeStateOf(CommissioningClient);
                if (state === undefined) {
                    continue;
                }
                const { addresses } = state;
                const isCommissioned = state.peerAddress !== undefined;

                // Shortcut for conditions we know no change is possible
                if (addresses === undefined || (isCommissioned && addresses.length === 1)) {
                    continue;
                }

                // Remove expired addresses
                let newAddresses = addresses.filter(addr => {
                    const exp = expirationOf(addr);
                    if (exp === undefined) {
                        return true;
                    }

                    return exp > now;
                });

                // Cull commissionable nodes that have expired
                if (!isCommissioned) {
                    if (!newAddresses?.length || (expirationOf(state) ?? 0) <= now) {
                        await node.delete();
                        continue;
                    }
                }

                // If the node is commissioned, do not remove the last address.  Instead keep the "least expired" addresses
                if (isCommissioned && addresses.length && !newAddresses.length) {
                    if (addresses.length === 1) {
                        continue;
                    }
                    const freshestExp = addresses.reduce((freshestExp, addr) => {
                        return Math.max(freshestExp, expirationOf(addr)!);
                    }, 0);

                    newAddresses = addresses.filter(addr => expirationOf(addr) === freshestExp);
                }

                // Apply new addresses if changed
                if (addresses.length !== newAddresses.length) {
                    await node.set({ commissioning: { addresses: newAddresses } });
                }
            }
        } finally {
            this.#expirationInterval = undefined;
        }
    }

    #instrumentBasicInformation(node: Endpoint, type: typeof BasicInformationClient) {
        if (!(node instanceof ClientNode)) {
            return;
        }

        node.eventsOf(type).leave?.on(({ fabricIndex }) => this.#onLeave(node, fabricIndex));
        node.eventsOf(type).shutDown?.on(() => this.#onShutdown(node));
        node.eventsOf(type).startUp?.on(() => this.#onStartUp(node));
        node.eventsOf(type).configurationVersion$Changed?.on(() => node.lifecycle.configurationVersionChanged.emit());
    }

    /**
     * Bridged endpoints carry their own `ConfigurationVersion` on `BridgedDeviceBasicInformation`; surface its changes
     * through the same endpoint lifecycle event used for the node's `BasicInformation`.
     */
    #instrumentBridgedConfigurationVersion(endpoint: Endpoint, type: typeof BridgedDeviceBasicInformationClient) {
        endpoint
            .eventsOf(type)
            .configurationVersion$Changed?.on(() => endpoint.lifecycle.configurationVersionChanged.emit());
    }

    #installIcdClient(node: Endpoint) {
        if (!(node instanceof ClientNode) || node.behaviors.has(IcdClient)) {
            return;
        }

        node.behaviors.inject(IcdClient);
    }

    #onLeave(node: ClientNode, fabricIndex: FabricIndex) {
        if (!node.lifecycle.isReady) {
            return;
        }

        this.#mutex.run(async () => {
            if (!node.lifecycle.isReady) {
                return;
            }

            const { fabrics: peerFabrics } = node.maybeStateOf(OperationalCredentialsClient) ?? {};
            const peerFabric = peerFabrics?.find(fabric => fabric.fabricIndex === fabricIndex);
            if (!peerFabric) {
                return;
            }

            const peerAddress = node.maybeStateOf(CommissioningClient)?.peerAddress;
            if (!peerAddress) {
                return;
            }

            const localFabrics = this.owner.env.get(FabricManager);
            const localFabric = localFabrics.forDescriptor(peerFabric);
            if (!localFabric || localFabric.fabricIndex !== peerAddress.fabricIndex) {
                return;
            }

            // Ignore leave events received during an initial subscription establishment as they may be stale events
            // from before the device was re-commissioned with the same identifier.
            // The reason is that we saw such cases and should prevent discarding a node directly after commissioning.
            // This solution still has some holes that could prevent removing nodes automatically, but best-effort
            // variant for now until we know how often that happens in practice.
            if (!node.act(agent => agent.get(NetworkClient).subscriptionActive)) {
                logger.info(
                    "Leave event for peer",
                    Diagnostic.strong(node.id),
                    " received without active subscription. Ignoring.",
                );
                return;
            }

            logger.notice("Peer", Diagnostic.strong(node.id), "has left the fabric");
            node.lifecycle.decommissioned.emit(LocalActorContext.ReadOnly);
            try {
                await node.env.maybeGet(Peer)?.disconnect(new PeerLeftError());
            } catch (error) {
                logger.warn(`Error force-closing sessions for ${node.id} on Leave:`, error);
            }
            await node.delete();
        });
    }

    async #onShutdown(node: ClientNode) {
        if (!node.lifecycle.isReady || !node.lifecycle.isOnline) {
            return;
        }

        // Ignore shutdown events received during initial subscription establishment as they may be stale events
        // from before the device was restarted.  This mirrors the same guard in #onLeave.
        if (!node.act(agent => agent.get(NetworkClient).subscriptionActive)) {
            logger.debug(
                "Shutdown event for peer",
                Diagnostic.strong(node.id),
                "received without active subscription. Ignoring.",
            );
            return;
        }

        const peerAddress = node.maybeStateOf(CommissioningClient)?.peerAddress;
        if (peerAddress !== undefined) {
            // Shutdown event means the device reboots, handle it like a peer loss and remove all sessions
            await this.owner.env.get(SessionManager).handlePeerShutdown(peerAddress);
        }
    }

    /**
     * Process startup events to detect node reboots in real-time. This allows us to terminate sessions
     * that would otherwise linger until timed out and cause interaction delays.
     */
    async #onStartUp(node: ClientNode) {
        // isOnline is checked because startUp arrives via an active subscription channel — the node
        // must be online for the subscription to deliver events.  isReady guards against races during
        // node initialization where the event might fire before the node is fully set up.
        if (!node.lifecycle.isReady || !node.lifecycle.isOnline) {
            return;
        }

        // Ignore startup events received during the initial subscription establishment
        // as they may be stale events from before the device was restarted.
        if (!node.act(agent => agent.get(NetworkClient).subscriptionActive)) {
            logger.debug(
                "Startup event for peer",
                Diagnostic.strong(node.id),
                "received without active subscription. Ignoring.",
            );
            return;
        }

        const peerAddress = node.maybeStateOf(CommissioningClient)?.peerAddress;
        if (peerAddress === undefined) {
            return;
        }

        // Use the current session's createdAt as asOf so it (and newer sessions) are preserved
        // while older sessions (from before the reboot) are closed.  If the currentSession is
        // undefined (no known session), asOf is undefined and handlePeerShutdown closes all sessions.
        const sessionManager = this.owner.env.get(SessionManager);
        await sessionManager.handlePeerShutdown(peerAddress, sessionManager.maybeSessionFor(peerAddress)?.createdAt);
    }
}

class Factory extends ClientNodeFactory {
    #owner: Peers;
    #groupIdCounter = 0;

    constructor(owner: Peers) {
        super();
        this.#owner = owner;
    }

    create(options: ClientNode.Options, peerAddress?: PeerAddress) {
        let node: ClientNode;
        if (peerAddress !== undefined && PeerAddress.isGroup(peerAddress)) {
            if (options.id === undefined) {
                options.id = `group${++this.#groupIdCounter}`;
            }
            node = new ClientGroup({
                ...options,
                owner: this.#owner.owner,
            });
        } else {
            if (options.id === undefined) {
                options.id = this.#owner.owner.env.get(ServerNodeStore).clientStores.allocateId();
            }
            node = new ClientNode({
                ...options,
                owner: this.#owner.owner,
            });
        }

        node.construction.start();
        return node;
    }

    find(descriptor: RemoteDescriptor) {
        for (const node of this.#owner) {
            // Skip nodes whose construction will not deliver a working backing.  Destroying/Destroyed close (or have
            // closed) the BehaviorBacking, which surfaces as "Datasource not yet initialized" the next time a caller
            // touches state.  Crashed never finished initializeDataSource.  Inactive/Initializing/Active are all
            // legitimate reuse targets — node.act will wait on construction.ready as needed.
            const status = node.construction.status;
            if (
                status === Lifecycle.Status.Destroying ||
                status === Lifecycle.Status.Destroyed ||
                status === Lifecycle.Status.Crashed
            ) {
                continue;
            }
            if (RemoteDescriptor.is(node.state.commissioning, descriptor)) {
                return node;
            }
        }
    }

    get nodes() {
        return this.#owner;
    }
}

function expirationOf<T extends { discoveredAt?: Timestamp; ttl?: Duration | number }>(
    lifespan: T,
): T extends { discoveredAt: Timestamp } ? Timestamp : Timestamp | undefined {
    if (lifespan.discoveredAt !== undefined) {
        return Timestamp(lifespan.discoveredAt + (Seconds(lifespan.ttl) ?? DEFAULT_TTL));
    }
    return undefined as unknown as Timestamp;
}