import 'dart:async'; import 'package:cloud_firestore/cloud_firestore.dart'; import '../../core/data/backend.dart'; import '../../core/data/shared_stream.dart'; import 'eclipse_data.dart'; /// 이클립스의 실 서버 구현. /// /// **요약 문서와 샤드를 함께 구독한다.** `fn-eclipse-aggregate` 는 1 분 주기인데 /// (Cloud Scheduler 최소 주기) 별첨 1 ① 시나리오 5 는 다른 단말의 참여가 내 화면에 /// 곧 보이기를 요구한다. 그래서 목표·마감 같은 정적 값은 요약 문서에서 읽고, /// **움직이는 값(방어량 · 참여 수)은 샤드 10 개를 직접 합산한다.** /// /// 침식률만은 서버가 적은 값을 그대로 쓴다. 화면이 다시 계산하면 단말마다 다른 /// 침식률이 뜨고, 그 순간 「방어했는데 수치가 안 맞는다」가 된다. /// /// **같은 팬덤·조회 범위의 구독은 공유한다.** HOME 배너·GALAXY 배너·행성 HUD가 /// 같은 집계를 보므로, 화면마다 `snapshots()` 를 열면 한 팬덤에 리스너가 화면 수만큼 /// 붙는다. [SharedStream]이 같은 원본을 나눠 주고 방어 화면은 마감 결과까지 별도로 받는다. class FirestoreEclipseRepository implements EclipseRepository { FirestoreEclipseRepository(this._backend); final Backend _backend; /// 내 소속 팬덤. 구독마다 다시 읽으면 화면 수만큼 왕복이 늘어 한 번만 읽는다. Future? _myFandomId; /// 팬덤별 공유 구독. 키가 빈 문자열이면 「내 팬덤」이다. final _watchers = >{}; final _tickers = >>{}; Future _fandomId(String? given) { if (given != null && given.isNotEmpty) return Future.value(given); return _myFandomId ??= _backend.db .doc('users/${_backend.requireUid}') .get() .then((snap) => (snap.data() ?? const {}).strOr('fandomId')); } @override Stream watch({ String? fandomId, bool includeResults = false, bool includeScheduled = false, }) => (_watchers['${fandomId ?? ''}#$includeResults#$includeScheduled'] ??= SharedStream( () => _openWatch(fandomId, includeResults, includeScheduled), )) .stream; @override Stream> liveEvents({String? fandomId, int limit = 20}) => (_tickers['${fandomId ?? ''}#$limit'] ??= SharedStream>( () => _openTicker(fandomId, limit), )).stream; Stream _openWatch( String? fandomId, bool includeResults, bool includeScheduled, ) => Stream.fromFuture(_fandomId(fandomId)).asyncExpand((id) { if (id.isEmpty) return Stream.value(null); return Stream.multi((out) { StreamSubscription>>? shardsSub; Timer? expiry; var generation = 0; final eventsSub = _backend.db .collection('fandoms/$id/eclipseRounds') .where('status', whereIn: [ if (includeScheduled) 'scheduled', 'active', if (includeResults) ...['ended', 'cancelled'], ]) .orderBy('startAt', descending: true) .limit(1) .snapshots(includeMetadataChanges: true) .where((s) => !s.metadata.isFromCache && !s.metadata.hasPendingWrites) .listen((events) { final current = ++generation; shardsSub?.cancel(); expiry?.cancel(); if (events.docs.isEmpty) { out.add(null); return; } final doc = events.docs.first; EclipseState? latest; final endsAt = doc.data()['endsAt']; if (endsAt is Timestamp) { final left = endsAt.toDate().difference(DateTime.now()); if (left <= Duration.zero && !includeResults) { out.add(null); return; } if (left > Duration.zero) { expiry = Timer(left, () { if (current == generation) out.add(includeResults ? latest : null); }); } } // asyncExpand는 무한 샤드 스트림이 끝날 때까지 다음 회차/종료를 // 막는다. 최신 요약이 올 때 이전 샤드 구독을 교체한다. shardsSub = doc.reference .collection('shards') .snapshots(includeMetadataChanges: true) .where( (s) => !s.metadata.isFromCache && !s.metadata.hasPendingWrites, ) .listen((shards) { if (current != generation) return; final state = _state( fandomId: id, eclipseId: doc.id, data: doc.data(), shards: shards.docs, ); latest = state; out.add(!includeResults && state.isExpired(DateTime.now()) ? null : state); }, onError: out.addError); }, onError: out.addError); out.onCancel = () async { generation++; expiry?.cancel(); await eventsSub.cancel(); await shardsSub?.cancel(); }; }); }); Stream> _openTicker(String? fandomId, int limit) => Stream.fromFuture(_fandomId(fandomId)).asyncExpand((id) { if (id.isEmpty) return Stream.value(const []); return _backend.db .collection('fandoms/$id/liveEvents') .orderBy('createdAt', descending: true) .limit(limit) .snapshots() .map( (snap) => [ for (final doc in snap.docs) _liveEvent(doc.id, doc.data()), ], ); }); @override Future defend({ required String fandomId, required String eclipseId, required String clientSeq, required int shakes, }) async { final data = await _backend.call('fnEclipseDefend', { 'fandomId': fandomId, 'eclipseId': eclipseId, 'clientSeq': clientSeq, 'shakes': shakes, }); return DefendResult( accepted: (data['accepted'] as num?)?.toInt() ?? 0, clamped: (data['clamped'] as num?)?.toInt() ?? 0, myTotal: (data['myTotal'] as num?)?.toInt() ?? 0, duplicate: data['duplicate'] == true, ); } EclipseState _state({ required String fandomId, required String eclipseId, required Map data, required List>> shards, }) { // 샤드 합. 샤드가 아직 없으면 요약 문서의 값으로 물러난다 — 첫 참여 전에는 // 샤드가 비어 있고, 그때 0 을 그리면 이미 모인 방어가 사라져 보인다. final live = shards.fold<(int, int)>((0, 0), (acc, doc) { final d = doc.data(); return (acc.$1 + d.intOr('qualifiedParticipants'), acc.$2 + d.intOr('defenseCount')); }); final endsAt = data['endsAt']; final startAt = data['startAt']; return EclipseState( fandomId: fandomId, eclipseId: eclipseId, fandomName: data.strOr('fandomName', fandomId), erosionRate: data.intOr('erosionRate', 100), defense: shards.isEmpty ? data.intOr('qualifiedParticipants') : live.$1, // 원장 C-25·C-26: 인정 횟수 이상인 고유 참여자 수 / DAU 기반 목표. targetDefense: data.intOr('target'), qualifyShakes: data.intOr('qualifyShakes', 60), defenseCount: shards.isEmpty ? data.intOr('defenseCount') : live.$2, defended: data.boolOr('defended') || data['result'] == 'success', result: switch (data['result']) { 'success' => EclipseResult.success, 'failed' => EclipseResult.failure, 'cancelled_error' => EclipseResult.cancelled, _ => null, }, startAt: startAt is Timestamp ? startAt.toDate() : null, warnAt: data['warnAt'] is Timestamp ? (data['warnAt'] as Timestamp).toDate() : null, scheduled: data['status'] == 'scheduled', endsAt: endsAt is Timestamp ? endsAt.toDate() : null, ); } LiveEvent _liveEvent(String id, Map data) => LiveEvent( id: id, // 서버가 닉네임을 비워 보내는 경우가 없도록 막아 두었으나, 옛 문서가 섞일 수 // 있어 화면 쪽에서도 기본값을 둔다. // 저장값 「팬」도 지금 언어의 「팬」으로(M-7). nickname: publicName(data['nickname']), defense: data.intOr('defense'), at: data.dateOr('createdAt', DateTime.now()), kind: LiveEventKind.of(data.strOr('kind')), priority: data.intOr('priority', 50), linkTo: data.strOr('linkTo'), ); }