import {
    AckPolicy,
    DebugEvents,
    Events,
    JetStreamClient,
    JetStreamManager,
    NatsConnection,
    RetentionPolicy,
    Stream,
    StreamInfo,
    connect,
    nanos,
    JSONCodec,
    Consumer,
} from 'nats';
import {
    Observable,
    Observer,
    ReplaySubject,
    catchError,
    combineLatest,
    delay,
    from,
    map,
    mergeMap,
    of,
    repeat,
    retry,
    switchMap,
    take,
    tap,
    timeout,
    toArray,
} from 'rxjs';
import { NatsClientConfig } from './nats-client-config.js';
import { NatsMessageRequest } from './nats-message-request.js';
import { NatsEventMessage } from './nats-event-message.js';
import { NatsMessageData } from './nats-message-data.js';

export class NatsClientService {
    private connection$ = new ReplaySubject<NatsConnection>(1);
    private jsManager$ = new ReplaySubject<JetStreamManager>(1);
    private jsClient$ = new ReplaySubject<JetStreamClient>(1);
    private streamInfoMap = new Map<string, StreamInfo>();
    private jsonCodec = JSONCodec();
    private readonly DEFAULT_TIMEOUT_MS = 10000;

    constructor(
        private options: NatsClientConfig,
        private logger: Console = console,
    ) {
        this.init();
    }

    async init() {
        this.initConnection(this.options)
            .pipe(
                switchMap((connection) =>
                    this.initJetstream(this.options, connection),
                ),
            )
            .subscribe();
    }

    // DATA QUEUE behaviour
    enqueueData(streamName: string, event: string, data: unknown) {
        this.jsClient$
            .pipe(
                timeout(this.DEFAULT_TIMEOUT_MS),
                take(1),
                switchMap((jsClient) =>
                    jsClient.publish(
                        `${streamName}.${event}`,
                        JSON.stringify(data),
                    ),
                ),
            )
            .subscribe();
    }

    dequeueData<T>(
        streamName: string,
        event: string = '*',
        consumerName: string = `durable_${streamName}_${event != '*' ? event : '_asterisk_'}`,
        clientId?: string,
    ): Observable<T> {
        return combineLatest({
            params: of({
                streamName,
                event,
                consumerName,
                clientId,
                streamOptions: this.options.streams.find(
                    (s) => s.name === streamName,
                ),
            }),
            jsClient: this.jsClient$,
            jsManager: this.jsManager$,
        }).pipe(
            switchMap(async ({ params, jsClient, jsManager }) => {
                const existingConsumer = await jsClient.consumers
                    .get(params.streamName, params.consumerName)
                    .catch(() => null);
                const consumerInfo = await (existingConsumer
                    ? jsManager.consumers.update(
                          params.streamName,
                          consumerName,
                          {
                              ack_wait: nanos(this.DEFAULT_TIMEOUT_MS),
                              max_deliver: 2,
                          },
                      )
                    : jsManager.consumers.add(params.streamName, {
                          durable_name: params.consumerName,
                          ack_policy: AckPolicy.Explicit,
                          max_ack_pending: 1,
                          filter_subjects: [
                              `${params.streamName}.${params.event}`,
                          ],
                      }));

                return jsClient.consumers
                    .get(consumerInfo.stream_name, consumerInfo.name)
                    .then((consumer) => ({ consumer, params }))
                    .catch((err) => ({ consumer: null as Consumer, params }));
            }),
            switchMap(({ consumer, params }) => {
                this.logger.log(
                    `new consumer loop for getting data from consumer ${params.consumerName} for event ${params.event} on stream ${params.streamName}, client id: ${params.clientId ?? 'N/A'}`,
                );
                return new Observable<T>((subs) => {
                    const consumerLoop$$ = of(consumer)
                        .pipe(
                            switchMap((consumer) => from(consumer.next())),
                            map((msg) => {
                                const value =
                                    msg?.data.length > 0 ? msg.json() : null;
                                msg?.ack();
                                return value as T;
                            }),
                            tap((value: T | null) =>
                                value ? subs.next(value) : null,
                            ),
                            repeat({
                                delay:
                                    params.streamOptions.consumerDelayMs ?? 300,
                            }),
                            catchError(() => {
                                this.logger.error(
                                    `error getting data from consumer ${params.consumerName} for event ${params.event} on stream ${params.streamName}, client id: ${params.clientId ?? 'N/A'}`,
                                );
                                return of(null);
                            }),
                        )
                        .subscribe({
                            error: (err) => subs.error(err),
                            complete: () => subs.complete(),
                        });

                    return () => consumerLoop$$.unsubscribe();
                });
            }),
        );
    }
    // ----- end QUEUE

    // MESSAGE behaviour
    requestMessage(
        subject: string,
        data: NatsMessageData,
        timeoutMs: number = this.DEFAULT_TIMEOUT_MS,
    ) {
        return this.connection$.pipe(
            timeout(this.DEFAULT_TIMEOUT_MS),
            take(1),
            switchMap((connection) =>
                connection.request(subject, this.jsonCodec.encode(data), {
                    timeout: timeoutMs,
                }),
            ),
            map((resp) => this.jsonCodec.decode(resp.data)),
        );
    }

    eventMessage(subject: string, data: NatsMessageData) {
        this.connection$
            .pipe(
                timeout(this.DEFAULT_TIMEOUT_MS),
                take(1),
                tap((connection) =>
                    connection.publish(subject, this.jsonCodec.encode(data)),
                ),
            )
            .subscribe();
    }

    subscribeToRequestMessage(subject: string, group: string) {
        return this.connection$.pipe(
            switchMap((connection) => {
                return new Observable((obs: Observer<NatsMessageRequest>) => {
                    const subscription = connection.subscribe(subject, {
                        queue: group,
                    });
                    (async (sub) => {
                        this.logger.debug(
                            `listening for ${sub.getSubject()} requests...`,
                        );
                        for await (const m of sub) {
                            obs.next(new NatsMessageRequest(m));
                        }
                        this.logger.debug(
                            `subscription ${sub.getSubject()} drained.`,
                        );
                    })(subscription);

                    return () => subscription.unsubscribe();
                });
            }),
        );
    }

    subscribeToEventMessage(subject: string, group: string) {
        return this.connection$.pipe(
            switchMap((connection) => {
                return new Observable((obs: Observer<NatsEventMessage>) => {
                    const subscription = connection.subscribe(subject, {
                        queue: group,
                    });
                    (async (sub) => {
                        this.logger.debug(
                            `listening for ${sub.getSubject()} requests...`,
                        );
                        for await (const m of sub) {
                            obs.next(new NatsEventMessage(m));
                        }
                        this.logger.debug(
                            `subscription ${sub.getSubject()} drained.`,
                        );
                    })(subscription);

                    return () => subscription.unsubscribe();
                });
            }),
        );
    }
    // ----- end MESSAGE

    // GENERIC
    consumerInfo(streamName: string) {
        return this.jsManager$.pipe(
            take(1),
            switchMap((jsManager) =>
                from(
                    jsManager.consumers.info(
                        streamName,
                        `durable_${streamName}`,
                    ),
                ),
            ),
        );
    }

    purgeStream(streamName: string) {
        return this.jsManager$.pipe(
            take(1),
            switchMap((jsManager) => from(jsManager.streams.purge(streamName))),
        );
    }

    streamInfo(streamName: string) {
        return this.jsManager$.pipe(
            take(1),
            switchMap((jsManager) =>
                from(jsManager.streams.get(streamName)).pipe(
                    switchMap((stream) => from(stream.info())),
                ),
            ),
        );
    }
    // ----- end GENERIC

    private async monitorClient(connection: NatsConnection) {
        connection.closed().then((error) => {
            this.init();
        });

        for await (const s of connection.status()) {
            switch (s.type) {
                case Events.Disconnect:
                    this.logger.log(`client disconnected - ${s.data}`);
                    break;
                case Events.LDM:
                    this.logger.log('client has been requested to reconnect');
                    break;
                case Events.Update:
                    this.logger.log(
                        `client received a cluster update - ${JSON.stringify(s.data)}`,
                    );
                    break;
                case Events.Reconnect:
                    this.logger.log(`client reconnected - ${s.data}`);
                    break;
                case Events.Error:
                    this.logger.log('client got a permissions error');
                    break;
                case DebugEvents.Reconnecting:
                    this.logger.log('client is attempting to reconnect');
                    break;
                case DebugEvents.StaleConnection:
                    this.logger.log('client has a stale connection');
                    break;
            }
        }
    }

    private initConnection(
        options: NatsClientConfig,
    ): Observable<NatsConnection> {
        return of(options).pipe(
            switchMap((options) => from(connect({ servers: options.servers }))),
            tap((connection) => {
                this.logger.log(
                    `Successfully connected to NATS server ${connection.info.host}`,
                );
                this.monitorClient(connection);
                this.connection$.next(connection);
            }),
            catchError((err) => {
                this.logger.error('Error connecting to NATS server!');
                throw err;
            }),
            retry({
                delay: (error: any, retryCount: number) => {
                    return of(retryCount).pipe(
                        tap(() =>
                            this.logger.warn(
                                `Waiting 30s before retrying to reconnect to NATS server...`,
                            ),
                        ),
                        delay(30000),
                        tap((retryCount) =>
                            this.logger.warn(
                                `Trying to reconnect to NATS server (attempt no.${retryCount})...`,
                            ),
                        ),
                    );
                },
                resetOnSuccess: true,
            }),
        );
    }

    private initJetstream(
        options: NatsClientConfig,
        connection: NatsConnection,
    ): Observable<{
        jsClient: JetStreamClient;
        jsManager: JetStreamManager;
        streamsInfo: StreamInfo[];
    }> {
        return of(connection).pipe(
            switchMap((connection) =>
                of(connection).pipe(
                    switchMap((connection) =>
                        combineLatest({
                            jsClient: of(connection.jetstream()).pipe(
                                catchError((err) => {
                                    this.logger.error(
                                        'Error getting the JETSTREAM client!',
                                    );
                                    throw err;
                                }),
                            ),
                            jsManager: from(connection.jetstreamManager()).pipe(
                                catchError((err) => {
                                    this.logger.error(
                                        'Error getting the JETSTREAM manager!',
                                    );
                                    throw err;
                                }),
                            ),
                        }),
                    ),
                    tap((jetstream) => {
                        this.jsClient$.next(jetstream.jsClient);
                        this.jsManager$.next(jetstream.jsManager);
                    }),
                    retry({
                        delay: (error: any, retryCount: number) => {
                            return of(retryCount).pipe(
                                tap(() =>
                                    this.logger.warn(
                                        `Waiting 30s before retrying to get JETSTREAM from NATS server...`,
                                    ),
                                ),
                                delay(30000),
                                tap((retryCount) =>
                                    this.logger.warn(
                                        `Trying to reconnect to get JETSTREAM from NATS server (attempt no.${retryCount})...`,
                                    ),
                                ),
                            );
                        },
                        resetOnSuccess: true,
                    }),
                ),
            ),
            switchMap((jetstream) =>
                this.initStreams(options, jetstream.jsManager).pipe(
                    map((streamsInfo) => ({ ...jetstream, streamsInfo })),
                ),
            ),
        );
    }

    private initStreams(
        options: NatsClientConfig,
        jsManager: JetStreamManager,
    ): Observable<StreamInfo[]> {
        return of({ options, jsManager }).pipe(
            switchMap(({ options, jsManager }) =>
                from(options.streams.map((i) => i.name)).pipe(
                    mergeMap((streamName) =>
                        from(
                            jsManager.streams.get(streamName).catch(() => null),
                        ).pipe(
                            map((stream: Stream) => ({
                                stream,
                                streamName,
                                jsManager,
                            })),
                        ),
                    ),
                    mergeMap(({ stream, streamName, jsManager }) => {
                        if (stream) return from(stream.info());
                        return from(
                            jsManager.streams.add({
                                name: streamName,
                                subjects: [`${streamName}.*`],
                                retention: RetentionPolicy.Interest,
                            }),
                        );
                    }),
                    tap((streamInfo) =>
                        this.streamInfoMap.set(
                            streamInfo.config.name,
                            streamInfo,
                        ),
                    ),
                    catchError((err) => {
                        this.logger.error('Error initializing NATS stream!');
                        throw err;
                    }),
                ),
            ),
            toArray(),
        );
    }
}