Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 29 additions & 0 deletions lib/features/earthquake/data/eew_realtime_source.dart
Original file line number Diff line number Diff line change
Expand Up @@ -7,9 +7,11 @@ import 'package:dpip/core/error/result.dart';
import 'package:dpip/core/logging/log.dart';
import 'package:dpip/core/network/api_exception.dart';
import 'package:dpip/core/network/sse_event.dart';
import 'package:dpip/core/realtime/app_time.dart';
import 'package:dpip/core/realtime/elapsed.dart';
import 'package:dpip/core/realtime/sse_realtime_source.dart';
import 'package:dpip/features/earthquake/domain/eew.dart';
import 'package:dpip/features/earthquake/domain/eew_feed.dart';
import 'package:flutter/foundation.dart' show listEquals;

/// The live EEW feed, streamed over Server-Sent Events (`/api/v2/eq/eew?sse=1`).
Expand Down Expand Up @@ -44,6 +46,11 @@ import 'package:flutter/foundation.dart' show listEquals;
/// of date, which is the truth — and the feed is `live` again only once the
/// stream answers, at which point polling stops at once and any poll still in
/// flight is discarded.
///
/// Every answer — streamed or polled — passes through [reduceEewFeed] on its
/// way out, so what the channel publishes is never a report older than one
/// already shown: the server a reconnect or a poll lands on may be a serial
/// behind the one before it.
class EewRealtimeSource extends SseRealtimeSource<List<Eew>> {
/// [connect] opens one EEW SSE connection — in production
/// `EarthquakeApi.openEewSse`; the source calls it again to reconnect.
Expand All @@ -54,20 +61,25 @@ class EewRealtimeSource extends SseRealtimeSource<List<Eew>> {
/// [cwaOnly] is read fresh on every [decode] and every poll (a closure, not
/// a captured bool) so toggling `EewCwaOnlySettings` takes effect on the
/// very next event instead of needing the connection to be rebuilt.
///
/// [now] stamps each snapshot for the reducer — the calibrated clock in
/// production, since it is compared with server-stamped instants.
EewRealtimeSource(
Stream<SseEvent> Function() connect, {
required bool Function() cwaOnly,
Future<List<dynamic>> Function()? poll,
Elapsed? elapsed,
Future<void> Function(Duration delay)? delay,
Random? random,
DateTime Function()? now,
}) : this._(
connect,
cwaOnly,
poll,
elapsed ?? SystemElapsed(),
delay,
random,
now ?? (() => AppTime.utc),
);

EewRealtimeSource._(
Expand All @@ -77,6 +89,7 @@ class EewRealtimeSource extends SseRealtimeSource<List<Eew>> {
this._monotonic,
Future<void> Function(Duration delay)? delay,
Random? random,
this._now,
) : super(
connect: connect,
label: 'eew',
Expand All @@ -103,6 +116,11 @@ class EewRealtimeSource extends SseRealtimeSource<List<Eew>> {
final bool Function() cwaOnly;
final Future<List<dynamic>> Function()? _poll;
final Elapsed _monotonic;
final DateTime Function() _now;

/// Survives reconnects, failovers and the switch to polling and back —
/// that is the point of it.
EewFeedState _feed = EewFeedState.initial;

/// When the stream started failing, or null while it answers.
Duration? _downSince;
Expand All @@ -125,8 +143,19 @@ class EewRealtimeSource extends SseRealtimeSource<List<Eew>> {
@override
bool get servingFallback => _servingFallback;

/// The answer, folded through [reduceEewFeed]. The channel calls this once
/// a second whether or not a new frame arrived, so the fold runs on the same
/// snapshot repeatedly; it is idempotent for an unchanged one.
@override
Future<Result<List<Eew>>> fetch() async {
final result = await _fetchUnreduced();
return result.map((frame) {
_feed = reduceEewFeed(_feed, frame, _now());
return _feed.visible;
});
}

Future<Result<List<Eew>>> _fetchUnreduced() async {
final streamed = await super.fetch();
final poll = _poll;
if (poll == null || _closed) return streamed;
Expand Down
76 changes: 76 additions & 0 deletions lib/features/earthquake/domain/eew.dart
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,11 @@ abstract class Eew with _$Eew {
@JsonKey(name: 'final', fromJson: boolishInt, toJson: intFromBool)
required bool isFinal,
@JsonKey(name: 'eq') required EewInfo info,

/// When this report was issued (Unix ms, server-stamped) — the top-level
/// `time`, as opposed to the origin in [EewInfo.time]. Null when a payload
/// leaves it out; the newest-report ordering then falls back to origin.
@JsonKey(name: 'time') int? reportTime,
}) = _Eew;

factory Eew.fromJson(Map<String, dynamic> json) => _$EewFromJson(json);
Expand All @@ -38,6 +43,77 @@ extension EewAgencyX on Eew {
bool get isJma => agency.toLowerCase() == 'jma';
}

/// Which earthquake a report belongs to.
extension EewIdentityX on Eew {
/// One event across all its reports: the publishing agency plus its own
/// event id — the same pair the server keys its list on, since two agencies
/// number their events independently. The agency is lower-cased because
/// nothing guarantees one feed spells it the same way every time.
String get eventKey => '${agency.toLowerCase()}:$id';
}

/// What a report is, read from [Eew.status].
enum EewReportKind {
/// 預警 — an ordinary report (`0`, and any code this app does not know).
forecast,

/// 警報 — the agency's public warning (`1`).
warning,

/// 取消 — the agency withdrew the event: it was not an earthquake, or not
/// the one the earlier reports described (`2` or `3`).
cancelled,
}

/// The meaning of the wire `status` integer.
///
/// The public JSON (`/api/v2/eq/eew`, GET and `?sse=1`) carries the
/// publishing agency's own code through untouched — the edge decodes it as a
/// single byte of the binary NATS message and copies it into the JSON — so
/// the cancel code depends on who encoded it:
///
/// | code | meaning | where it comes from |
/// |---|---|---|
/// | `0` | 預警 | every agency |
/// | `1` | 警報 | CWA's public warning |
/// | `2` | 取消 | the event server's own model (`0 normal, 1 alert, 2 cancel`), which is how the `nied` feed's cancel arrives |
/// | `3` | 取消 | the CWA feed, and what every ExpTech client keys its cancel styling on |
///
/// No feed uses `2` or `3` for anything else, so both read as a cancel.
/// [Eew.isFinal] (`final`) is separate: `1` marks the agency's last report
/// for the event, which says nothing about whether it was cancelled.
///
/// An unknown code reads as [EewReportKind.forecast], never as a cancel: a
/// misread code must leave an alert on screen, not hide one.
extension EewStatusX on Eew {
EewReportKind get kind => switch (status) {
1 => EewReportKind.warning,
2 || 3 => EewReportKind.cancelled,
_ => EewReportKind.forecast,
};

/// Whether the agency withdrew this event. A cancelled report must never
/// drive an alert: no countdown, no local estimate, no wave rings, no
/// speech — only a "cancelled" notice until it leaves the list.
bool get isCancelled => kind == EewReportKind.cancelled;

/// Whether this is the agency's public 警報 rather than a 預警.
bool get isWarning => kind == EewReportKind.warning;
}

/// The cancel filter every consumer of an alert list needs.
extension EewListX on List<Eew> {
/// The alerts that still describe an earthquake in progress — everything
/// but the cancelled ones, in their original order.
List<Eew> get uncancelled => [
for (final alert in this)
if (!alert.isCancelled) alert,
];

/// Whether any alert still describes an earthquake in progress.
bool get hasUncancelled => any((alert) => !alert.isCancelled);
}

/// The earthquake parameters carried by an [Eew].
@freezed
abstract class EewInfo with _$EewInfo {
Expand Down
49 changes: 29 additions & 20 deletions lib/features/earthquake/domain/eew.freezed.dart

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

Loading
Loading