// Copyright (c) Cratis. All rights reserved.
// Licensed under the MIT license. See LICENSE file in the project root for full license information.

import type {
    AppendedEvent,
    ReadModelChangeset as ContractReadModelChangeset,
    ReadModelObserverType
} from '@cratis/chronicle.contracts';
import {
    ReadModelObserverType as ContractReadModelObserverType,
    ReadModelChangeType as ContractReadModelChangeType
} from '@cratis/chronicle.contracts';
import type { Constructor } from '@cratis/fundamentals';
import { JsonSerializer } from '@cratis/fundamentals';
import { IClientArtifactsProvider } from '../artifacts/index.js';
import { ChronicleConnection } from '../connection/index.js';
import type { ConnectionLifecycle } from '../connection/ConnectionLifecycle.js';
import { fromContractsGuid } from '../connection/Guid.js';
import { ensureQuerySuccess } from '../connection/callResults.js';
import { EventSequenceId } from '../eventSequences/EventSequenceId.js';
import { getProjectionMetadata } from '../projections/declarative/projection.js';
import { hasFromEventMetadata } from '../projections/modelBound/fromEvent.js';
import { hasModelBoundProperties } from '../types/TypeDiscoverer.js';
import { isPassive } from '../projections/modelBound/passive.js';
import { getReducerMetadata } from '../reducers/reducer.js';
import { JsonSchemaGenerator, type JsonSchema } from '../schemas/index.js';
import { WellKnownSinks } from '../sinks/index.js';
import { getReadModelMetadata, getReadModelId } from './readModel.js';
import { buildReadModelDefinition } from './buildReadModelDefinition.js';
import { assertUniqueReadModelIds } from './assertUniqueReadModelIds.js';
import { rootReadModelTypes } from './rootReadModelTypes.js';
import type { IMaterializedReadModels } from './IMaterializedReadModels.js';
import { MaterializedReadModels } from './MaterializedReadModels.js';
import { ReadModelSubjectResolver } from './ReadModelSubjectResolver.js';
import { deserializeReadModel } from './deserializeReadModel.js';
import type { ReadModelNamingPolicy } from './ReadModelNamingPolicy.js';
import type { IReadModels } from './IReadModels.js';
import type { IReadModelWatcher } from './IReadModelWatcher.js';
import { ReadModelWatcher } from './ReadModelWatcher.js';
import type { ReadModelChangeset } from './ReadModelChangeset.js';
import type { ReadModelWatchOptions } from './ReadModelWatchOptions.js';
import { ReadModelChangeType } from './ReadModelChangeType.js';
import type { ReadModelSnapshot } from './ReadModelSnapshot.js';

const unlimitedEventCount = BigInt('18446744073709551615');

interface ResolvedReadModel {
    readonly type: Constructor;
    readonly identifier: string;
    readonly eventSequenceId: string;
    readonly observerType: ReadModelObserverType;
    readonly observerIdentifier: string;
    readonly schema: string;
    readonly isActive: boolean;
    readonly isModelBound?: boolean;
}

/**
 * Implements {@link IReadModels} by working with the Chronicle read-model gRPC service.
 */
export class ReadModels implements IReadModels {
    readonly materialized: IMaterializedReadModels;
    private readonly _watchCancellation = new AbortController();

    constructor(
        private readonly _eventStore: string,
        private readonly _namespace: string,
        private readonly _connection: ChronicleConnection,
        private readonly _clientArtifacts: IClientArtifactsProvider,
        private readonly _defaultSinkTypeId: string,
        private readonly _isModelBoundProjectionRegistered?: (readModelType: Constructor) => boolean,
        private readonly _readModelNamingPolicy?: ReadModelNamingPolicy,
        private readonly _lifecycle?: ConnectionLifecycle
    ) {
        this.materialized = new MaterializedReadModels(_eventStore, _namespace, _connection);
    }

    /** @inheritdoc */
    async register<TReadModel>(readModelType?: Constructor<TReadModel>): Promise<void> {
        const readModels = this.resolveReadModels(readModelType);
        if (readModels.length === 0) {
            return;
        }

        await this._connection.readModels.registerMany({
            EventStore: this._eventStore,
            Owner: 1,
            Source: 1,
            ReadModels: readModels.map(readModel => this.toDefinition(readModel))
        });
    }

    /** @inheritdoc */
    async getInstanceById<TReadModel>(readModelType: Constructor<TReadModel>, key: string, sessionId?: string): Promise<TReadModel> {
        const readModel = this.resolveReadModel(readModelType);
        const response = await this._connection.readModels.getInstanceByKey({
            EventStore: this._eventStore,
            Namespace: this._namespace,
            ReadModelIdentifier: readModel.identifier,
            EventSequenceId: readModel.eventSequenceId,
            ReadModelKey: key,
            SessionId: sessionId ?? ''
        });

        const instance = this.deserializeReadModel(readModelType, response.ReadModel);

        if (readModel.observerType === ContractReadModelObserverType.Reducer && this.schemaHasComplianceMetadata(readModel.schema)) {
            return this.release(readModelType, instance);
        }

        return instance;
    }

    /** @inheritdoc */
    async findInstanceById<TReadModel>(readModelType: Constructor<TReadModel>, key: string, sessionId?: string): Promise<TReadModel | null> {
        const readModel = this.resolveReadModel(readModelType);
        const response = await this._connection.readModels.getInstanceByKey({
            EventStore: this._eventStore,
            Namespace: this._namespace,
            ReadModelIdentifier: readModel.identifier,
            EventSequenceId: readModel.eventSequenceId,
            ReadModelKey: key,
            SessionId: sessionId ?? ''
        });

        if (!response.ReadModel || response.ReadModel.trim() === 'null') {
            return null;
        }
        const instance = this.deserializeReadModel(readModelType, response.ReadModel);

        if (readModel.observerType === ContractReadModelObserverType.Reducer && this.schemaHasComplianceMetadata(readModel.schema)) {
            return this.release(readModelType, instance);
        }

        return instance;
    }

    /** @inheritdoc */
    async getInstances<TReadModel>(readModelType: Constructor<TReadModel>, eventCount: bigint = unlimitedEventCount): Promise<TReadModel[]> {
        const readModel = this.resolveReadModel(readModelType);
        const response = await this._connection.readModels.getAllInstances({
            EventStore: this._eventStore,
            Namespace: this._namespace,
            ReadModelIdentifier: readModel.identifier,
            EventSequenceId: readModel.eventSequenceId,
            EventCount: eventCount
        });

        const instances = response.Instances.map(instance => this.deserializeReadModel(readModelType, instance));

        if (readModel.observerType === ContractReadModelObserverType.Reducer && this.schemaHasComplianceMetadata(readModel.schema)) {
            return this.releaseMany(readModelType, instances);
        }

        return instances;
    }

    /** @inheritdoc */
    async getSnapshotsById<TReadModel>(readModelType: Constructor<TReadModel>, key: string): Promise<ReadModelSnapshot<TReadModel>[]> {
        const readModel = this.resolveReadModel(readModelType);
        const response = ensureQuerySuccess('get read model snapshots', await this._connection.readModelExplorer.allSnapshotsForReadModel({
            EventStore: this._eventStore,
            Namespace: this._namespace,
            ReadModel: readModel.identifier,
            EventSequenceId: readModel.eventSequenceId,
            ReadModelKey: key,
            Grouping: ''
        }));

        const snapshots: ReadModelSnapshot<TReadModel>[] = response.map(snapshot => ({
            readModel: this.deserializeReadModel(readModelType, snapshot.Instance),
            events: (snapshot.Events ?? []) as AppendedEvent[],
            occurred: snapshot.Occurred?.Value ? new Date(snapshot.Occurred.Value) : undefined,
            correlationId: snapshot.CorrelationId
        }));

        if (readModel.observerType === ContractReadModelObserverType.Reducer && this.schemaHasComplianceMetadata(readModel.schema)) {
            return this.releaseSnapshotInstances(readModelType, snapshots);
        }

        return snapshots;
    }

    /** @inheritdoc */
    async *watch<TReadModel>(readModelType: Constructor<TReadModel>): AsyncIterable<ReadModelChangeset<TReadModel>> {
        const readModel = this.resolveReadModel(readModelType);

        for await (const changeset of this._connection.readModels.watch({
            EventStore: this._eventStore,
            Namespace: this._namespace,
            ReadModelIdentifier: readModel.identifier,
            EventSequenceId: readModel.eventSequenceId
        })) {
            if (changeset.Subscribed) {
                continue;
            }

            yield await this.toChangeset(readModelType, readModel, changeset);
        }
    }

    /** @inheritdoc */
    createWatcher<TReadModel>(readModelType: Constructor<TReadModel>, options?: ReadModelWatchOptions): IReadModelWatcher<TReadModel> {
        const readModel = this.resolveReadModel(readModelType);
        const signal = AbortSignal.any(options?.signal
            ? [options.signal, this._watchCancellation.signal]
            : [this._watchCancellation.signal]);
        return new ReadModelWatcher(
            streamSignal => this._connection.readModels.watch({
                EventStore: this._eventStore,
                Namespace: this._namespace,
                ReadModelIdentifier: readModel.identifier,
                EventSequenceId: readModel.eventSequenceId
            }, { signal: streamSignal }),
            changeset => this.toChangeset(readModelType, readModel, changeset),
            signal,
            this._lifecycle,
            options
        );
    }

    /** Stops all watchers created through createWatcher for this event store. */
    dispose(): void {
        this._watchCancellation.abort();
    }

    /** @inheritdoc */
    async dehydrateSession<TReadModel>(sessionId: string, readModelType: Constructor<TReadModel>, key: string): Promise<void> {
        const readModel = this.resolveReadModel(readModelType);
        await this._connection.readModels.dehydrateSession({
            EventStore: this._eventStore,
            Namespace: this._namespace,
            ReadModelIdentifier: readModel.identifier,
            EventSequenceId: readModel.eventSequenceId,
            ReadModelKey: key,
            SessionId: sessionId
        });
    }

    /** @inheritdoc */
    async release<TReadModel>(readModelType: Constructor<TReadModel>, instance: TReadModel): Promise<TReadModel> {
        const readModel = this.resolveReadModel(readModelType);
        const schema = this.getReadModelSchema(readModelType, readModel.identifier);
        const payload = JsonSerializer.serialize(instance);
        const subject = this.extractSubject(readModelType, instance);

        const response = await this._connection.compliance.release({
            EventStore: this._eventStore,
            Namespace: this._namespace,
            Subject: subject,
            Schema: schema,
            Payload: payload
        });

        if (response.HasError) {
            throw new Error(`Failed to release PII: ${response.Error}`);
        }

        return this.deserializeReadModel(readModelType, response.Payload);
    }

    /** @inheritdoc */
    async releaseMany<TReadModel>(readModelType: Constructor<TReadModel>, instances: TReadModel[]): Promise<TReadModel[]> {
        const releasePromises = instances.map(instance => this.release(readModelType, instance));
        return Promise.all(releasePromises);
    }

    /** @inheritdoc */
    async releaseDocument<TReadModel>(readModelType: Constructor<TReadModel>, document: Readonly<Record<string, unknown>>): Promise<Record<string, unknown>> {
        const readModel = this.resolveReadModel(readModelType);
        const schema = JSON.parse(this.getReadModelSchema(readModelType, readModel.identifier)) as JsonSchema;
        const properties = schema.properties ?? {};
        const stored = this.copyStoredDocument(document);
        const storedDefaultSubject = stored.__subject;
        const defaultSubject = this.isDocumentSubject(storedDefaultSubject) ? storedDefaultSubject : undefined;
        const storedSubjects = this.isDocumentObject(stored.__subjects) ? stored.__subjects : {};
        const subjects = new Map<string, string>();

        if (storedDefaultSubject !== undefined && storedDefaultSubject !== null && typeof storedDefaultSubject !== 'string') {
            throw new Error('Stored read model document has an invalid default subject.');
        }
        for (const [property, subject] of Object.entries(storedSubjects)) {
            if (property === '__subject' || property === '__subjects') {
                throw new Error('Stored read model document has an unknown subject property.');
            }
            // The kernel's JSON subject map ignores null/empty strings, but rejects non-string values.
            if (subject !== null && typeof subject !== 'string') {
                throw new Error('Stored read model document has an invalid property subject.');
            }
            // Cleared or removed properties can leave stale mappings in stored documents.
            if (!Object.hasOwn(stored, property)) {
                continue;
            }
            if (!Object.hasOwn(properties, property)) {
                throw new Error('Stored read model document has an unknown subject property.');
            }
            if (this.isDocumentSubject(subject)) {
                subjects.set(property, subject);
            }
        }

        const groups = new Map<string, string[]>();
        const releasedEntries: [string, unknown][] = [];
        for (const [property, value] of Object.entries(stored)) {
            // Only schema-declared data is returned, never storage bookkeeping.
            if (property === '__subject' || property === '__subjects' || !Object.hasOwn(properties, property)) {
                continue;
            }
            const subject = subjects.get(property) ?? defaultSubject;
            if (subject === undefined) {
                releasedEntries.push([property, value]);
                continue;
            }
            const group = groups.get(subject) ?? [];
            group.push(property);
            groups.set(subject, group);
        }

        for (const [subject, propertyNames] of groups) {
            const payload = Object.fromEntries(propertyNames.map(property => [property, stored[property]]));
            const groupSchema: JsonSchema = {
                ...schema,
                properties: Object.fromEntries(propertyNames.map(property => [property, properties[property]])),
                required: schema.required?.filter(property => propertyNames.includes(property))
            };
            const released = await this.releaseDocumentProperties(subject, groupSchema, payload);
            const releasedProperties = Object.keys(released);
            if (releasedProperties.length !== propertyNames.length ||
                releasedProperties.some(property => !Object.hasOwn(payload, property))) {
                throw new Error('Failed to release stored read model document: missing or conflicting properties.');
            }
            releasedEntries.push(...Object.entries(released));
        }

        // Publish only after every subject succeeded. Entire top-level subtrees belong to one group.
        return Object.fromEntries(releasedEntries);
    }

    private async releaseDocumentProperties(subject: string, schema: JsonSchema, payload: Record<string, unknown>): Promise<Record<string, unknown>> {
        try {
            const response = await this._connection.compliance.release({
                EventStore: this._eventStore,
                Namespace: this._namespace,
                Subject: subject,
                Schema: JSON.stringify(schema),
                Payload: JSON.stringify(payload)
            });
            if (response.HasError !== false || typeof response.Payload !== 'string') {
                throw new Error();
            }
            const released: unknown = JSON.parse(response.Payload);
            if (!this.isDocumentObject(released)) {
                throw new Error();
            }
            return released;
        } catch {
            // Kernel, transport and JSON parser errors can contain PII; do not expose them as causes.
            throw new Error('Failed to release PII in stored read model document.');
        }
    }

    private copyStoredDocument(document: Readonly<Record<string, unknown>>): Record<string, unknown> {
        try {
            if (!this.isDocumentObject(document)) {
                throw new Error();
            }
            const copy: unknown = JSON.parse(JSON.stringify(document));
            if (!this.isDocumentObject(copy)) {
                throw new Error();
            }
            return copy;
        } catch {
            throw new Error('Stored read model document must be a JSON object.');
        }
    }

    private isDocumentObject(value: unknown): value is Record<string, unknown> {
        return value !== null && typeof value === 'object' && !Array.isArray(value);
    }

    private isDocumentSubject(value: unknown): value is string {
        return typeof value === 'string' && value.length > 0;
    }

    private async toChangeset<TReadModel>(readModelType: Constructor<TReadModel>, readModel: ResolvedReadModel, changeset: ContractReadModelChangeset): Promise<ReadModelChangeset<TReadModel>> {
        const instance = this.deserializeReadModel(readModelType, changeset.ReadModel);
        const occurred = changeset.Occurred?.Value ? new Date(changeset.Occurred.Value) : undefined;
        const requiresRelease = !changeset.Removed && readModel.observerType === ContractReadModelObserverType.Reducer &&
            this.schemaHasComplianceMetadata(readModel.schema);
        return {
            namespace: changeset.Namespace,
            key: changeset.ModelKey,
            readModel: requiresRelease ? await this.release(readModelType, instance) : instance,
            removed: changeset.Removed,
            changeType: changeset.ChangeType === ContractReadModelChangeType.Added ? ReadModelChangeType.Added :
                changeset.ChangeType === ContractReadModelChangeType.Removed ? ReadModelChangeType.Removed : ReadModelChangeType.Modified,
            changeContext: {
                eventStore: this._eventStore,
                namespace: changeset.Namespace,
                sequenceNumber: changeset.EventSequenceNumber,
                correlationId: changeset.CorrelationId ? fromContractsGuid(changeset.CorrelationId).toString() : undefined,
                occurred: occurred && !Number.isNaN(occurred.getTime()) ? occurred : undefined
            }
        };
    }

    private async releaseSnapshotInstances<TReadModel>(readModelType: Constructor<TReadModel>, snapshots: ReadModelSnapshot<TReadModel>[]): Promise<ReadModelSnapshot<TReadModel>[]> {
        return Promise.all(
            snapshots.map(async snapshot => ({
                ...snapshot,
                readModel: await this.release(readModelType, snapshot.readModel)
            }))
        );
    }

    private resolveReadModels<TReadModel>(readModelType?: Constructor<TReadModel>): ResolvedReadModel[] {
        const rootTypes = rootReadModelTypes(this._clientArtifacts);
        assertUniqueReadModelIds(rootTypes);
        const resolved = new Map<string, ResolvedReadModel>();

        for (const projectionType of this._clientArtifacts.projections) {
            const metadata = getProjectionMetadata(projectionType);
            if (!metadata?.readModelType) {
                continue;
            }

            if (readModelType && metadata.readModelType !== readModelType) {
                continue;
            }

            const identifier = getReadModelId(metadata.readModelType);
            this.addResolved(resolved, {
                type: metadata.readModelType,
                identifier,
                eventSequenceId: metadata.eventSequenceId ?? EventSequenceId.eventLog.value,
                observerType: ContractReadModelObserverType.Projection,
                observerIdentifier: metadata.id.value,
                schema: this.getReadModelSchema(metadata.readModelType, identifier),
                isActive: !isPassive(metadata.readModelType)
            });
        }

        for (const modelBoundType of rootTypes) {
            if (!hasFromEventMetadata(modelBoundType) && !hasModelBoundProperties(modelBoundType)) {
                continue;
            }

            if (readModelType && modelBoundType !== readModelType) {
                continue;
            }

            const identifier = getReadModelId(modelBoundType);
            this.addResolved(resolved, {
                type: modelBoundType,
                identifier,
                eventSequenceId: EventSequenceId.eventLog.value,
                observerType: ContractReadModelObserverType.Projection,
                observerIdentifier: identifier,
                schema: this.getReadModelSchema(modelBoundType, identifier),
                isActive: !isPassive(modelBoundType),
                isModelBound: true
            });
        }

        for (const reducerType of this._clientArtifacts.reducers) {
            const metadata = getReducerMetadata(reducerType);
            if (!metadata?.readModel) {
                continue;
            }

            if (readModelType && metadata.readModel !== readModelType) {
                continue;
            }

            const identifier = getReadModelId(metadata.readModel);
            this.addResolved(resolved, {
                type: metadata.readModel,
                identifier,
                eventSequenceId: metadata.eventSequenceId ?? EventSequenceId.eventLog.value,
                observerType: ContractReadModelObserverType.Reducer,
                observerIdentifier: metadata.id.value,
                schema: this.getReadModelSchema(metadata.readModel, identifier),
                isActive: true
            });
        }

        return Array.from(resolved.values());
    }

    private addResolved(models: Map<string, ResolvedReadModel>, model: ResolvedReadModel): void {
        const existing = models.get(model.identifier);
        if (existing && (existing.type !== model.type || existing.observerType !== model.observerType ||
            existing.observerIdentifier !== model.observerIdentifier)) {
            throw new Error(`Read model id '${model.identifier}' has multiple observers or model types.`);
        }
        models.set(model.identifier, model);
    }

    private resolveReadModel<TReadModel>(readModelType: Constructor<TReadModel>): ResolvedReadModel {
        const [resolved] = this.resolveReadModels(readModelType);
        const neverRegistered = resolved?.isModelBound && this._isModelBoundProjectionRegistered && !this._isModelBoundProjectionRegistered(readModelType);
        if (!resolved || neverRegistered) {
            throw new Error(hasModelBoundProperties(readModelType)
                ? `Unknown read model '${readModelType.name}'. It has model-bound property mappings but was not registered when the event store was created. Export the class from a module matched by discoveryPatterns, or register it explicitly before creating the event store.`
                : `Unknown read model '${readModelType.name}'. Make sure it is discoverable through a projection, reducer, or model-bound mapping.`);
        }
        return resolved;
    }


    private toDefinition(readModel: ResolvedReadModel) {
        return buildReadModelDefinition({
            identifier: readModel.identifier,
            type: readModel.type,
            schema: readModel.schema,
            // Passive read models never write to a materialized sink, so they register with the
            // None sink and the kernel resolves them via immediate projection instead of an empty sink.
            sinkTypeId: readModel.isActive ? this._defaultSinkTypeId : WellKnownSinks.None,
            observerType: readModel.observerType,
            observerIdentifier: readModel.observerIdentifier,
            namingPolicy: this._readModelNamingPolicy
        });
    }

    private getReadModelSchema(readModelType: Constructor, identifier: string): string {
        const metadata = getReadModelMetadata(readModelType);
        if (metadata) {
            return JSON.stringify(metadata.schema);
        }

        return JSON.stringify(JsonSchemaGenerator.generate(readModelType) ?? JsonSchemaGenerator.createEmptySchema(identifier));
    }

    private deserializeReadModel<TReadModel>(readModelType: Constructor<TReadModel>, json: string): TReadModel {
        return deserializeReadModel(readModelType, json);
    }

    private schemaHasComplianceMetadata(schema: string): boolean {
        try {
            const parsed = JSON.parse(schema) as Record<string, unknown>;
            const properties = parsed.properties as Record<string, Record<string, unknown>> | undefined;
            if (!properties) {
                return false;
            }
            return Object.values(properties).some(
                property => Array.isArray(property.compliance) && (property.compliance as unknown[]).length > 0
            );
        } catch {
            return false;
        }
    }

    private extractSubject<TReadModel>(readModelType: Constructor<TReadModel>, instance: TReadModel): string {
        const subject = ReadModelSubjectResolver.resolveFrom(readModelType, instance);
        if (subject !== undefined) {
            return subject;
        }
        throw new Error('Read model instance must have a property decorated with @subject() or an "id" property to serve as the subject for PII release');
    }
}