gongxue-base/apps/server/src/attendance/attendance-import.service.ts

import { Injectable, Logger } from '@nestjs/common';
import { InjectRepository } from '@nestjs/typeorm';
import { Repository, In } from 'typeorm';
import { Subject, Observable } from 'rxjs';
import {
  DingAttendanceRaw,
  Student,
  UserDingMapping,
} from '../entities';
import { DingTalkService, DingTalkAttendanceResult } from '../integration/dingtalk.service';
import { AttendanceService } from './attendance.service';
import type { ImportProgressEvent, ImportResult } from './dto/dingtalk-import.dto';

/**
 * Service for importing DingTalk attendance data into the system.
 *
 * ## Stream Architecture
 * Uses RxJS Subjects to emit progress events during the import pipeline:
 *   fetch → parse → deduplicate → save → auto-match
 *
 * Progress is exposed as an Observable so the controller can relay it via SSE.
 * Only ONE import can run at a time (guarded by `isRunning`).
 */
@Injectable()
export class AttendanceImportService {
  private readonly logger = new Logger(AttendanceImportService.name);

  /** RxJS Subject emitting live progress during import */
  private progressSubject = new Subject<ImportProgressEvent>();
  private isRunning = false;
  constructor(
    @InjectRepository(DingAttendanceRaw)
    private readonly dingRawRepo: Repository<DingAttendanceRaw>,
    @InjectRepository(Student)
    private readonly studentRepo: Repository<Student>,
    @InjectRepository(UserDingMapping)
    private readonly userDingMappingRepo: Repository<UserDingMapping>,
    private readonly dingTalkService: DingTalkService,
    private readonly attendanceService: AttendanceService,
  ) {}

  /**
   * Expose progress as a read-only Observable for SSE.
   */
  get progress$(): Observable<ImportProgressEvent> {
    return this.progressSubject.asObservable();
  }

  /**
   * Whether an import is currently in progress.
   */
  get running(): boolean {
    return this.isRunning;
  }

  /**
   * Run the DingTalk attendance import pipeline.
   *
   * Pipeline stages:
   * 1. Fetch attendance results from DingTalk (paginated)
   * 2. Parse and validate each record
   * 3. Deduplicate by `dingId` (unique in DB)
   * 4. Batch-save to `ding_attendance_raw`
   * 5. Optionally auto-match to students by name
   */
  async importFromDingTalk(params: {
    startDate: string;
    endDate: string;
    userIds?: string[];
    autoMatch?: boolean;
  }): Promise<ImportResult> {
    if (this.isRunning) {
      throw new Error('An import is already in progress');
    }

    const startedAt = Date.now();
    this.isRunning = true;

    // Safety timeout: auto-reset isRunning after 30 minutes in case of
    // an unhandled exception that bypasses the finally block (extremely rare).
    const SAFETY_TIMEOUT_MS = 30 * 60 * 1000;
    const safetyTimer = setTimeout(() => {
      if (this.isRunning) {
        this.logger.error('Import safety timeout triggered — force-resetting isRunning');
        this.isRunning = false;
      }
    }, SAFETY_TIMEOUT_MS);

    const errors: string[] = [];
    let imported = 0;
    let skipped = 0;
    let matched = 0;

    try {
      // Stage 1: Fetch
      this.emit('fetching', 0, 0, 'Fetching attendance results from DingTalk...');
      const rawResults = await this.fetchAllPages(params);
      const total = rawResults.length;
      this.emit('fetching', total, total, `Fetched ${total} raw attendance records`);

      // Stage 2: Parse & deduplicate
      this.emit('parsing', 0, total, `Parsing ${total} records...`);
      const existingDingIds = await this.getExistingDingIds(rawResults);
      const newRecords = rawResults.filter((r) => !existingDingIds.has(r.checkId));
      skipped = rawResults.length - newRecords.length;
      this.emit('parsing', newRecords.length, total, `${newRecords.length} new records, ${skipped} duplicates skipped`);

      if (newRecords.length === 0) {
        this.emit('complete', imported + skipped, total, 'Nothing new to import');
        return { success: true, imported, skipped, matched, errors, duration: Date.now() - startedAt };
      }

      // Stage 3: Batch save
      this.emit('saving', 0, newRecords.length, `Saving ${newRecords.length} records...`);
      const batchSize = 100;
      for (let i = 0; i < newRecords.length; i += batchSize) {
        const batch = newRecords.slice(i, i + batchSize);
        const entities = batch.map((r) => this.mapToEntity(r));
        try {
          await this.dingRawRepo.save(entities, { chunk: 50 });
          imported += entities.length;
          this.emit('saving', imported, newRecords.length, `Saved ${imported}/${newRecords.length}`);
        } catch (err: unknown) {
          const msg = err instanceof Error ? err.message : String(err);
          errors.push(`Batch save error at offset ${i}: ${msg}`);
          this.logger.error(`Batch save error: ${msg}`);
        }
      }

      // Stage 4: Auto-match (optional)
      if (params.autoMatch && imported > 0) {
        this.emit('matching', 0, imported, 'Auto-matching records to students...');
        matched = await this.autoMatchUnmatched();
        this.emit('matching', matched, imported, `Matched ${matched} records to students`);
      }

      const duration = Date.now() - startedAt;
      this.emit('complete', imported, rawResults.length, `Import complete: ${imported} new, ${skipped} skipped, ${matched} matched (${duration}ms)`);
      this.logger.log(`DingTalk attendance import done: ${imported} imported, ${skipped} skipped, ${matched} matched`);
      return { success: true, imported, skipped, matched, errors, duration };
    } catch (err: unknown) {
      const msg = err instanceof Error ? err.message : String(err);
      errors.push(msg);
      this.emit('error', imported, 0, `Import failed: ${msg}`, msg);
      this.logger.error(`DingTalk attendance import failed: ${msg}`);
      return { success: false, imported, skipped, matched, errors, duration: Date.now() - startedAt };
    } finally {
      clearTimeout(safetyTimer);
      this.isRunning = false;
    }
  }

  /**
   * Paginate through DingTalk attendance API.
   * The DingTalk API returns max 50 records per page.
   */
  private async fetchAllPages(params: {
    startDate: string;
    endDate: string;
    userIds?: string[];
  }): Promise<DingTalkAttendanceResult[]> {
    const allResults: DingTalkAttendanceResult[] = [];
    const pageSize = 50;
    let offset = 0;
    let hasMore = true;

    while (hasMore) {
      const batch = await this.dingTalkService.fetchAttendanceResults({
        startDate: params.startDate,
        endDate: params.endDate,
        userIds: params.userIds,
        offset,
        limit: pageSize,
      });

      if (batch.length === 0) {
        hasMore = false;
      } else {
        allResults.push(...batch);
        offset += batch.length;
        this.emit('fetching', allResults.length, allResults.length + (batch.length < pageSize ? 0 : pageSize), `Fetched ${allResults.length} records...`);
        // If last page was smaller than pageSize, we're done
        if (batch.length < pageSize) hasMore = false;
      }
    }

    return allResults;
  }

  /**
   * Query which dingIds already exist to skip duplicates.
   */
  private async getExistingDingIds(
    results: DingTalkAttendanceResult[],
  ): Promise<Set<string>> {
    const dingIds = results.map((r) => r.checkId).filter(Boolean);
    if (dingIds.length === 0) return new Set();

    const existing = await this.dingRawRepo.find({
      where: { dingId: In(dingIds) },
      select: ['dingId'],
    });
    return new Set(existing.map((e) => e.dingId));
  }

  /**
   * Map a DingTalk API result to a DingAttendanceRaw entity.
   */
  private mapToEntity(r: DingTalkAttendanceResult): DingAttendanceRaw {
    const entity = new DingAttendanceRaw();
    entity.dingUserId = r.userId;
    entity.userName = ''; // Will be filled from the result if available
    entity.attendanceDate = r.workDate;
    entity.dingId = r.checkId;
    entity.attendanceType = r.checkType || 'OnDuty';
    entity.timeResult = r.timeResult;
    entity.locationResult = r.locationResult || '';

    // Parse check-in/out times
    if (r.actualCheckTime) {
      const dt = new Date(r.actualCheckTime);
      if (!isNaN(dt.getTime())) {
        if (r.checkType === 'OnDuty') {
          entity.checkInTime = dt;
        } else {
          entity.checkOutTime = dt;
        }
      }
    }

    entity.matchStatus = '未处理';
    entity.rawData = JSON.stringify(r);
    return entity;
  }

  /**
   * Auto-match unmatched records to students via the dingUserId → userId mapping chain.
   *
   * Delegates to {@link AttendanceService.autoMatchDingRecords} to avoid duplicate logic.
   */
  private async autoMatchUnmatched(): Promise<number> {
    const result = await this.attendanceService.autoMatchDingRecords();
    return result.matched;
  }

  /**
   * Emit a progress event to the stream.
   */
  private emit(
    phase: ImportProgressEvent['phase'],
    current: number,
    total: number,
    message: string,
    error?: string,
  ): void {
    this.progressSubject.next({ phase, current, total, message, error });
  }
}