import {
  Injectable,
  Logger,
  type OnModuleDestroy,
  type OnModuleInit,
} from '@nestjs/common';
import { InjectRepository } from '@nestjs/typeorm';
import { Repository } from 'typeorm';
import { Subject, merge, interval, type Observable } from 'rxjs';
import { filter, map } from 'rxjs/operators';
import Redis from 'ioredis';
import {
  paginated,
  resolvePage,
  type PaginatedResult,
} from '../../common/pagination/pagination';
import {
  Notification,
  type NotificationType,
} from './entities/notification.entity';

// 알림 응답(원시) — 표시 문구/아이콘/톤은 프론트가 type+payload 로 조립
export interface NotificationResponse {
  id: string;
  type: NotificationType;
  payload: Record<string, unknown>;
  read: boolean;
  createdAt: string;
}

// 목록 응답 — 페이지네이션 + 미읽음 수 동봉
export type NotificationListResult = PaginatedResult<NotificationResponse> & {
  unreadCount: number;
};

// 알림 적재 파라미터
export interface NotifyParams {
  recipientIds: string[]; // 수신자 User uuid 목록
  actorId: string | null; // 행위자(수신자에서 제외 — 자기 행위는 알리지 않음)
  type: NotificationType;
  payload?: Record<string, unknown>;
}

// SSE 스트림에 흘려보내는 내부 이벤트(수신자 id + 알림 본문)
interface StreamEvent {
  userId: string;
  data: NotificationResponse;
}

// 알림 비즈니스 로직 — 다른 도메인 서비스가 notify() 로 적재하고,
// 사용자별 수신함을 읽거나(SSE) 읽음 처리한다. 적재는 베스트 에포트(본 흐름을 막지 않음).
//
// 실시간 전송:
// - REDIS_HOST 설정 시 **Redis pub/sub** — notify 가 채널에 발행하고 모든 인스턴스가 구독해
//   각자의 로컬 Subject 로 전달한다. → 다중 인스턴스(수평 확장)에서도 실시간 도달.
// - 미설정 시 **인메모리** 직접 전달(단일 인스턴스). 어느 모드든 SSE 구독은 로컬 stream$ 만 본다.
@Injectable()
export class NotificationService implements OnModuleInit, OnModuleDestroy {
  private readonly logger = new Logger(NotificationService.name);
  // 이 인스턴스의 SSE 구독자에게 전달하는 로컬 Subject(구독 시 userId 로 필터)
  private readonly stream$ = new Subject<StreamEvent>();
  // Redis pub/sub 연결(미설정 시 null → 인메모리 모드). 구독 연결은 일반 명령 불가라 분리한다.
  private publisher: Redis | null = null;
  private subscriber: Redis | null = null;
  // 실시간 알림 채널명
  private static readonly CHANNEL = 'relay:notifications';

  constructor(
    @InjectRepository(Notification)
    private readonly notificationRepo: Repository<Notification>,
  ) {}

  // 기동 시 Redis pub/sub 연결 — REDIS_HOST 있을 때만. 실패해도 인메모리로 폴백.
  onModuleInit(): void {
    const host = process.env.REDIS_HOST;
    if (!host) {
      this.logger.log(
        '알림 실시간 전송: 인메모리 모드(단일 인스턴스) — REDIS_HOST 미설정',
      );
      return;
    }
    const options = {
      host,
      port: Number(process.env.REDIS_PORT) || 6379,
      password: process.env.REDIS_PASSWORD || undefined,
      // 장애가 unhandled 로 죽지 않도록 — 무한 재시도(백오프)하며 로그만 남김
      maxRetriesPerRequest: null,
    };
    this.publisher = new Redis(options);
    this.subscriber = new Redis(options);
    this.publisher.on('error', (e: Error) =>
      this.logger.warn(`Redis publisher 오류(무시): ${e.message}`),
    );
    this.subscriber.on('error', (e: Error) =>
      this.logger.warn(`Redis subscriber 오류(무시): ${e.message}`),
    );
    // 채널 구독 — 자기 인스턴스가 발행한 것도 여기서 받아 로컬 stream$ 로 단일 전달
    this.subscriber
      .subscribe(NotificationService.CHANNEL)
      .catch((e: unknown) =>
        this.logger.warn(
          `Redis 구독 실패: ${e instanceof Error ? e.message : String(e)}`,
        ),
      );
    this.subscriber.on('message', (channel: string, message: string) => {
      if (channel !== NotificationService.CHANNEL) return;
      try {
        this.stream$.next(JSON.parse(message) as StreamEvent);
      } catch {
        // 파싱 실패 무시
      }
    });
    this.logger.log('알림 실시간 전송: Redis pub/sub 모드(다중 인스턴스 지원)');
  }

  // 종료 시 Redis 연결 정리
  onModuleDestroy(): void {
    this.publisher?.disconnect();
    this.subscriber?.disconnect();
  }

  // 실시간 이벤트 전달 — Redis 모드면 발행(모든 인스턴스가 구독으로 수신),
  // 인메모리 모드면 로컬 Subject 로 직접 전달.
  private dispatch(event: StreamEvent): void {
    if (this.publisher) {
      this.publisher
        .publish(NotificationService.CHANNEL, JSON.stringify(event))
        .catch((e: unknown) => {
          // 발행 실패 시 최소한 이 인스턴스 구독자에게는 직접 전달(폴백)
          this.logger.warn(
            `Redis 발행 실패, 로컬 폴백: ${e instanceof Error ? e.message : String(e)}`,
          );
          this.stream$.next(event);
        });
    } else {
      this.stream$.next(event);
    }
  }

  // 알림 적재 + 실시간 전송 — 실패해도 호출부 흐름을 깨지 않도록 내부에서 흡수
  async notify(params: NotifyParams): Promise<void> {
    try {
      // 중복 제거 + 행위자/빈 값 제외
      const recipients = [...new Set(params.recipientIds)].filter(
        (id) => id && id !== params.actorId,
      );
      if (recipients.length === 0) return;

      const entities = recipients.map((recipientId) =>
        this.notificationRepo.create({
          recipient: { id: recipientId } as Notification['recipient'],
          type: params.type,
          payload: params.payload ?? {},
          read: false,
        }),
      );
      const saved = await this.notificationRepo.save(entities);

      // 저장 성공분을 각 수신자에게 실시간 전달(Redis 발행 또는 로컬)
      for (let i = 0; i < saved.length; i++) {
        this.dispatch({
          userId: recipients[i],
          data: this.toResponse(saved[i]),
        });
      }
    } catch (err) {
      this.logger.error(
        `알림 적재 실패: ${params.type}`,
        err instanceof Error ? err.stack : String(err),
      );
    }
  }

  // 사용자 수신함 — 최신순 페이지네이션 + 미읽음 수
  async listFor(
    userId: string,
    page?: number,
    size?: number,
  ): Promise<NotificationListResult> {
    const pg = resolvePage(page, size);
    const [rows, total] = await this.notificationRepo.findAndCount({
      where: { recipient: { id: userId } },
      order: { createdAt: 'DESC' },
      skip: pg.skip,
      take: pg.take,
    });
    const unreadCount = await this.notificationRepo.count({
      where: { recipient: { id: userId }, read: false },
    });
    return {
      ...paginated(
        rows.map((n) => this.toResponse(n)),
        total,
        pg,
      ),
      unreadCount,
    };
  }

  // 단건 읽음 처리(본인 알림만) — 처리 후 미읽음 수 반환
  async markRead(userId: string, id: string): Promise<{ unreadCount: number }> {
    await this.notificationRepo.update(
      { id, recipient: { id: userId } },
      { read: true },
    );
    return { unreadCount: await this.unreadCount(userId) };
  }

  // 전체 읽음 처리(본인 알림) — 처리 후 미읽음 수(0) 반환
  async markAllRead(userId: string): Promise<{ unreadCount: number }> {
    await this.notificationRepo.update(
      { recipient: { id: userId }, read: false },
      { read: true },
    );
    return { unreadCount: 0 };
  }

  // 사용자별 실시간 스트림(SSE) — 본인 알림만 필터 + 주기적 하트비트(프록시 idle 차단 방지)
  streamFor(userId: string): Observable<MessageEvent> {
    const events = this.stream$.asObservable().pipe(
      filter((e) => e.userId === userId),
      // 기본('message') 이벤트로 전송 → 프론트 EventSource.onmessage 가 수신
      map((e) => ({ data: e.data }) as MessageEvent),
    );
    // 'ping' 은 명명 이벤트라 onmessage 를 트리거하지 않는다(연결 유지 전용)
    const heartbeat = interval(25_000).pipe(
      map(() => ({ type: 'ping', data: {} }) as MessageEvent),
    );
    return merge(events, heartbeat);
  }

  // --- 내부 헬퍼 ---

  private async unreadCount(userId: string): Promise<number> {
    return this.notificationRepo.count({
      where: { recipient: { id: userId }, read: false },
    });
  }

  private toResponse(n: Notification): NotificationResponse {
    return {
      id: n.id,
      type: n.type,
      payload: n.payload ?? {},
      read: n.read,
      createdAt: n.createdAt.toISOString(),
    };
  }
}
