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

import { Behavior } from "#behavior/Behavior.js";
import { ClusterBehavior } from "#behavior/cluster/ClusterBehavior.js";
import type { ClusterState } from "#behavior/cluster/ClusterState.js";
import { ActionContext } from "#behavior/context/ActionContext.js";
import { ValueSupervisor } from "#behavior/supervision/ValueSupervisor.js";
import { Endpoint } from "#endpoint/Endpoint.js";
import {
    asError,
    BasicSet,
    Diagnostic,
    Environment,
    Environmental,
    InternalError,
    Logger,
    ObserverGroup,
    serialize,
} from "@matter/general";
import { type AttributeModel, DataModelPath } from "@matter/model";
import {
    AccessControl,
    assertRemoteActor,
    Fabric,
    FabricManager,
    hasRemoteActor,
    PeerAddress,
    Subject,
    Val,
} from "@matter/protocol";
import {
    AtomicAttributeStatus,
    AttributeId,
    ClusterId,
    type ClusterTyping,
    NodeId,
    Status,
    StatusResponse,
    StatusResponseError,
} from "@matter/types";
import { Thermostat } from "@matter/types/clusters/thermostat";
import { AtomicWriteState } from "./AtomicWriteState.js";

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

/**
 * Handles atomic write handling according to Matter definitions.
 * The implementation tries to be generic, but is currently only used by the Thermostat cluster, so the atomic write
 * types are imported from there.
 *
 * The logic requires that the cluster behavior implements the following additional events as "pure Observable()" events,
 * because the current implementation logic requires error thrown by the event handlers to signal validation failures to
 * be thrown back to te emitter. This is not the case for official state events.
 * * `${attributeName}$AtomicChanging` - emitted when an attribute is changed as part of an atomic write, before the value
 *   is actually changed. Receives the new value, the old value and the action context as parameters.
 * * `${attributeName}$AtomicChanged` - emitted on commit once every attribute of the atomic write is staged in the
 *   transaction, so validation spanning attributes sees their pending values. Receives the new value, the old value
 *   and the action context as parameters.
 *
 * TODO: Move out of thermostat behavior into a more generic behavior handler once used by other clusters too. Then we
 *  also need to adjust how it is handled.
 *  Proper solution might be to add the handling of the atomic Request command on interaction level and leave the
 *  transaction open until it is rolled back or committed. This might have side effects on other parts of the system though.
 *  So lets do that later when we have more clusters using it.
 */
export class AtomicWriteHandler {
    #observers = new ObserverGroup();
    #pendingWrites = new BasicSet<AtomicWriteState>();

    constructor(fabricManager: FabricManager) {
        this.#observers.on(fabricManager.events.deleting, fabric => this.#handleFabricRemoval(fabric));
    }

    static [Environmental.create](env: Environment) {
        const instance = new AtomicWriteHandler(env.get(FabricManager));
        env.set(AtomicWriteHandler, instance);
        return instance;
    }

    close() {
        this.#observers.close();
        for (const writeState of Array.from(this.#pendingWrites)) {
            writeState.close();
        }
    }

    /**
     * Validates the parts of a request that every request type shares and that fail the whole command, so they come
     * before any per-attribute status.
     *
     * @see {@link MatterSpecification.v161.Core} § 7.15.6.4
     */
    #validateRequest(
        { attributeRequests }: Thermostat.AtomicRequest,
        context: ActionContext,
        endpoint: Endpoint,
        cluster: ClusterBehavior.Type,
    ) {
        // Ensure we have a valid peer and so also associated fabric
        const peerAddress = this.#assertValidPeer(context);

        if (attributeRequests.length === 0) {
            throw new StatusResponse.InvalidCommandError("No attribute requests provided");
        }
        const attributes = new Map<AttributeId, AttributeModel>();
        for (const attr of attributeRequests) {
            const model = cluster.schema.conformant.attributes(attr);
            if (model === undefined || endpoint.stateOf(cluster.id)[attr] === undefined) {
                throw new StatusResponse.InvalidCommandError(`Attribute ${attr} not supported by cluster`);
            }
            if (attributes.has(attr)) {
                throw new StatusResponse.InvalidCommandError("Duplicate attribute in attribute requests");
            }
            attributes.set(attr, model);
        }

        return { peerAddress, attributes };
    }

    /**
     * Returns the atomic write state a CommitWrite or RollbackWrite request addresses.
     */
    #assertExistingState<B extends Behavior.Type>(
        request: Thermostat.AtomicRequest,
        context: ActionContext,
        endpoint: Endpoint,
        cluster: B,
    ) {
        if (!ClusterBehavior.is(cluster)) {
            throw new InternalError("Cluster behavior expected for atomic write handler");
        }
        const { peerAddress } = this.#validateRequest(request, context, endpoint, cluster);
        const { attributeRequests } = request;

        const existingState = this.#pendingWriteForPeer(peerAddress, endpoint, cluster.cluster.id);
        if (existingState === undefined) {
            throw new StatusResponse.InvalidInStateError("No atomic write in progress for this peer");
        }
        if (
            existingState.attributeRequests.length !== attributeRequests.length ||
            !existingState.attributeRequests.every(attr => attributeRequests.includes(attr))
        ) {
            throw new StatusResponse.InvalidInStateError("Attribute requests do not match existing atomic write");
        }
        return existingState;
    }

    /**
     * Implements the begin write logic for an atomic write.
     */
    beginWrite(
        request: Thermostat.AtomicRequest,
        context: ActionContext,
        endpoint: Endpoint,
        cluster: Behavior.Type,
    ): Thermostat.AtomicResponse {
        if (!hasRemoteActor(context)) {
            throw new StatusResponse.InvalidCommandError("AtomicRequest requires a remote actor");
        }
        if (!ClusterBehavior.is(cluster)) {
            throw new InternalError("Cluster behavior expected for atomic write handler");
        }

        const { peerAddress, attributes } = this.#validateRequest(request, context, endpoint, cluster);

        // §7.15.6.4.1 step 1: reject at command level if this peer already holds an atomic write on the cluster/endpoint
        // for any attributes, before per-attribute processing (matches CHIP's InAtomicWrite gate).
        if (this.#pendingWriteForPeer(peerAddress, endpoint, cluster.cluster.id)) {
            throw new StatusResponse.InvalidInStateError(
                "An atomic write is already in progress for this peer on this cluster and endpoint",
            );
        }

        const { attributeRequests, timeout } = request;
        if (timeout === undefined) {
            throw new StatusResponse.InvalidCommandError("Timeout missing for BeginWrite request");
        }

        let commandStatusCode = Status.Success;
        const attributeStatus = [...attributes].map(([attr, attributeModel]) => {
            let statusCode = Status.Success;
            const { writeLevel } = cluster.supervisor.get(attributeModel).access.limits;
            const location = {
                path: DataModelPath.none,
                endpoint: endpoint.number,
                cluster: cluster.cluster.id,
                owningFabric: context.fabric,
            };
            if (context.authorityAt(writeLevel, location) !== AccessControl.Authority.Granted) {
                statusCode = Status.UnsupportedAccess;
            } else if (!attributeModel.effectiveQuality.atomic) {
                statusCode = Status.InvalidCommand;
            } else if (this.#pendingWriteStateForAttribute(endpoint, cluster, attr) !== undefined) {
                statusCode = Status.Busy;
            }

            if (statusCode !== Status.Success) {
                commandStatusCode = Status.Failure;
            }
            return {
                attributeId: attr,
                statusCode,
            };
        });

        if (commandStatusCode !== Status.Success) {
            return { statusCode: commandStatusCode, attributeStatus };
        }

        const initialValues: Val.Struct = {};
        for (const attr of attributeRequests) {
            initialValues[attr] = endpoint.stateOf(cluster.id)[attr];
        }
        const state = new AtomicWriteState(
            peerAddress,
            endpoint,
            cluster.cluster.id,
            attributeRequests,
            timeout,
            new Map([...attributes].map(([attr, { propertyName }]) => [attr, propertyName])),
            initialValues,
        );
        this.#pendingWrites.add(state);
        state.closed.once(() => void this.#pendingWrites.delete(state));
        state.start();
        logger.debug("Added atomic write state:", state);

        return {
            statusCode: commandStatusCode,
            attributeStatus,
            timeout: state.timeout,
        };
    }

    /**
     * Handles writing a value for an attribute as part of an ongoing atomic write.
     * It uses the *$AtomicChanging* event to trigger validation of the partial write.
     */
    writeAttribute(
        context: ValueSupervisor.Session,
        endpoint: Endpoint,
        cluster: ClusterBehavior.Type,
        attribute: AttributeId,
        value: unknown,
    ) {
        const state = this.#assertPendingWriteForAttributeAndPeer(context, endpoint, cluster, attribute);
        const attributeName = state.attributeNames.get(attribute)!;
        logger.debug(`Writing pending value for attribute ${attributeName}, ${attribute} in atomic write`, value);
        // TODO currently we only handle this one changing, so checking other state within the event potentially use
        //  older values. We need to tweak the state for a complete solution. But ok for now!
        endpoint
            .eventsOf(cluster.id)
            [`${attributeName}$AtomicChanging`]?.emit(
                value,
                state.pendingAttributeValues[attribute] !== undefined
                    ? state.pendingAttributeValues[attribute]
                    : state.initialValues[attribute],
                context,
            );
        state.pendingAttributeValues[attribute] = value;
        logger.debug("Atomic write state after current write:", state);
    }

    /**
     * Implements the commit logic for an atomic write.
     */
    async commitWrite<B extends Behavior.Type>(
        request: Thermostat.AtomicRequest,
        context: ActionContext,
        endpoint: Endpoint,
        cluster: B,
        clusterState: ClusterState.Type<ClusterTyping, B>,
    ): Promise<Thermostat.AtomicResponse> {
        const state = this.#assertExistingState(request, context, endpoint, cluster);
        const events = endpoint.eventsOf(cluster.id);

        // §7.15.6.4.2: the pending writes apply as one unit, so every attribute is staged in this one transaction and
        // any failure discards all of them
        const attributeStatus = request.attributeRequests.map((attributeId): AtomicAttributeStatus => ({
            attributeId,
            statusCode: Status.Success,
        }));
        const staged = new Array<{ status: AtomicAttributeStatus; name: string; value: unknown; oldValue: unknown }>();

        try {
            for (const status of attributeStatus) {
                const value = state.pendingAttributeValues[status.attributeId];
                if (value === undefined) {
                    continue;
                }
                const name = state.attributeNames.get(status.attributeId)!;
                const oldValue = endpoint.stateOf(cluster.id)[status.attributeId];
                try {
                    events[`${name}$AtomicChanging`]?.emit(value, oldValue, context);
                    (clusterState as any)[status.attributeId] = value;
                    staged.push({ status, name, value, oldValue });
                } catch (error) {
                    status.statusCode = this.#statusOf(error, `staging attribute ${name}`);
                }
            }

            for (const { status, name, value, oldValue } of staged) {
                try {
                    events[`${name}$AtomicChanged`]?.emit(value, oldValue, context);
                } catch (error) {
                    status.statusCode = this.#statusOf(error, `validating attribute ${name}`);
                }
            }

            if (attributeStatus.some(({ statusCode }) => statusCode !== Status.Success)) {
                await context.transaction?.rollback();
            } else {
                try {
                    await context.transaction?.commit();
                } catch (error) {
                    // A failure while committing cannot be traced to one attribute, so it stands for each of them
                    const statusCode = this.#statusOf(error, "committing");
                    for (const status of attributeStatus) {
                        status.statusCode = statusCode;
                    }
                }
            }
        } finally {
            state.close();
        }

        // The command reports a generic failure whatever went wrong; the attributes' own statuses carry the reason, and
        // certification reads them that way
        return {
            statusCode: attributeStatus.every(({ statusCode }) => statusCode === Status.Success)
                ? Status.Success
                : Status.Failure,
            attributeStatus,
        };
    }

    /**
     * The status an attribute reports for an error raised while committing it.  A refused value is routine for the
     * peer; an error that carries no Matter status is a defect, which fails only this atomic write.
     */
    #statusOf(error: unknown, step: string) {
        const statusCode = StatusResponseError.of(error)?.code;
        if (statusCode !== undefined) {
            logger.debug(`Atomic write commit failed ${step}:`, Diagnostic.errorMessage(asError(error)));
            return statusCode;
        }
        logger.warn(`Atomic write commit failed unexpectedly ${step}:`, error);
        return Status.Failure;
    }

    /**
     * Implements the rollback logic for an atomic write.
     */
    rollbackWrite(
        request: Thermostat.AtomicRequest,
        context: ActionContext,
        endpoint: Endpoint,
        cluster: Behavior.Type,
    ): Thermostat.AtomicResponse {
        const state = this.#assertExistingState(request, context, endpoint, cluster);
        state.close();
        return {
            statusCode: Status.Success,
            attributeStatus: state.attributeRequests.map(attr => ({
                attributeId: attr,
                statusCode: Status.Success,
            })),
        };
    }

    /**
     * Handles fabric removal by closing all pending atomic write states for peers on the removed fabric.
     */
    #handleFabricRemoval(fabric: Fabric) {
        const fabricIndex = fabric.fabricIndex;
        for (const writeState of Array.from(this.#pendingWrites)) {
            if (writeState.peerAddress.fabricIndex === fabricIndex) {
                logger.debug(
                    `Closing atomic write state for peer ${writeState.peerAddress.toString()} on endpoint ${writeState.endpoint.id} due to fabric removal`,
                );
                writeState.close();
            }
        }
    }

    /**
     * Returns the atomic write state this peer holds on the given cluster/endpoint, if any.
     */
    #pendingWriteForPeer(peerAddress: PeerAddress, endpoint: Endpoint, clusterId: ClusterId) {
        return this.#pendingWrites.find(
            s =>
                PeerAddress.is(s.peerAddress, peerAddress) &&
                s.endpoint.number === endpoint.number &&
                s.clusterId === clusterId,
        );
    }

    /**
     * Returns the pending write state for the given attribute, if any.
     */
    #pendingWriteStateForAttribute(endpoint: Endpoint, cluster: ClusterBehavior.Type, attribute: AttributeId) {
        const writeStates = this.#pendingWrites.filter(
            s => s.endpoint.number === endpoint.number && s.clusterId === cluster.cluster.id,
        );
        if (writeStates.length === 0) {
            return undefined;
        }
        const attrWriteStates = writeStates.filter(({ attributeRequests }) => attributeRequests.includes(attribute));
        if (attrWriteStates.length === 0) {
            return undefined;
        }
        if (attrWriteStates.length > 1) {
            throw new InternalError("Multiple atomic write states found for the same attribute. Should never happen");
        }
        return attrWriteStates[0];
    }

    /**
     * Returns the pending value for the given attribute and peer, if any.
     */
    pendingValueForAttributeAndPeer(
        context: ValueSupervisor.Session,
        endpoint: Endpoint,
        cluster: ClusterBehavior.Type,
        attribute: AttributeId,
    ) {
        const peerAddress = this.#derivePeerAddress(context);
        if (peerAddress === undefined) {
            // No valid peer address could be derived from the session
            return undefined;
        }
        const attrWriteState = this.#pendingWriteStateForAttribute(endpoint, cluster, attribute);
        if (attrWriteState === undefined) {
            // No pending write for this attribute
            return undefined;
        }
        if (!PeerAddress.is(attrWriteState.peerAddress, peerAddress)) {
            // Pending state s for an other peer
            return undefined;
        }
        logger.debug(
            `Found pending value for attribute ${attribute} for peer ${peerAddress.nodeId}`,
            serialize(attrWriteState.pendingAttributeValues[attribute]),
        );
        return attrWriteState.pendingAttributeValues[attribute];
    }

    #assertPendingWriteForAttributeAndPeer(
        session: ValueSupervisor.Session,
        endpoint: Endpoint,
        cluster: ClusterBehavior.Type,
        attribute: AttributeId,
    ) {
        const attrWriteState = this.#pendingWriteStateForAttribute(endpoint, cluster, attribute);
        if (attrWriteState === undefined) {
            throw new StatusResponse.InvalidInStateError("There is no atomic write in progress for this attribute");
        }
        const peerAddress = this.#derivePeerAddress(session);
        if (peerAddress === undefined) {
            throw new StatusResponse.InvalidInStateError("There is no atomic write in progress for this peer");
        }
        if (!PeerAddress.is(attrWriteState.peerAddress, peerAddress)) {
            // CHIP and TC_TSTAT_4_2 require BUSY here, not the §7.15.3 INVALID_IN_STATE, when another peer holds the
            // atomic write (INVALID_IN_STATE is reserved for when no atomic write is open on the attribute)
            throw new StatusResponse.BusyError("Attribute is part of an atomic write in progress for a different peer");
        }
        return attrWriteState;
    }

    #derivePeerAddress(session: ValueSupervisor.Session) {
        if (
            hasRemoteActor(session) &&
            Subject.isNode(session.subject) &&
            NodeId.isOperationalNodeId(session.subject.id)
        ) {
            return PeerAddress({ fabricIndex: session.fabric, nodeId: NodeId(session.subject.id) });
        }
    }

    #assertValidPeer(context: ActionContext) {
        assertRemoteActor(context); // Also validate that we have a remote actor for processing the command

        const peerAddress = this.#derivePeerAddress(context);
        if (!context.session.associatedFabric || peerAddress === undefined) {
            throw new StatusResponse.InvalidCommandError("AtomicRequest requires an operational session");
        }
        return peerAddress;
    }
}