Schedule backend: AniList client, AnimeFlow client, schedule matching

lib/src/schedule/ for the 1.0 Schedule tab (UI and docs come with the
schedule round):
- AniListClient: airingSchedules in a window, 50 per page, adult/music
  dropped; memory 10 min + disk 1 h cache, one Retry-After wait on 429,
  stale cache on failure; real User-Agent (AniList 403s some defaults).
- AnimeFlowClient: the full app API (me, schedule, series, follow,
  search, requests incl. approve/reject, users/scopes) with the user's
  Jellyfin token, only for a Jellyfin on AnimeFlow's parent domain;
  ETag/304, typed error codes, backoff on nginx's HTML 429, /me cached
  at most 60 s so scope changes show.
- matchSchedule + ScheduleRepository: AniList airings with AnimeFlow
  status by (episode anilist_id, episode), then absolute number, then
  air day; degrades to AniList-only when AnimeFlow fails.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
foxyzandClaude Opus 5.5 committed 2026-10-08 23:47:33 +03:00
1 parent c76fee70dc
commit caa0765890
7 files changed
+2072

No files matched your search

+328
View File
@@ -0,0 +1,328 @@
import 'dart:async';
import 'dart:convert';
import 'dart:io';
import 'dart:ui' show Color;
import 'package:path_provider/path_provider.dart';
import 'package:sync_relay_core/core.dart';
import '../update/updater.dart' show appVersion;
// AniList airing schedule (public GraphQL, no login): every episode airing
// in a time range, with covers, banners and popularity.
//
// AniList allows ~90 requests/min (sometimes degraded to 30), so the raw
// pages are cached: in memory for 10 min (tab switches) and on disk for an
// hour (restarts), keyed by range start. A 429 waits out Retry-After once;
// if AniList still refuses (or is unreachable), stale cached data is
// returned with its fetch time.
final _log = createLogger('anilist');
const aniListDefaultUrl = 'https://graphql.anilist.co';
/// One episode airing, as AniList announces it.
class Airing {
final int anilistId;
/// English title, else romaji.
final String title;
final String titleRomaji;
final int episode;
/// UTC.
final DateTime airingAt;
final String? cover;
final String? banner;
/// The cover's dominant color, if AniList has one.
final Color? color;
final int popularity;
const Airing({
required this.anilistId,
required this.title,
required this.titleRomaji,
required this.episode,
required this.airingAt,
this.cover,
this.banner,
this.color,
this.popularity = 0,
});
/// Null for adult entries, music videos and malformed rows.
static Airing? fromJson(Object? j) {
if (j is! Map) return null;
final m = j['media'], at = j['airingAt'], ep = j['episode'];
if (m is! Map || at is! int || ep is! int) return null;
final id = m['id'];
if (id is! int || m['isAdult'] == true || m['format'] == 'MUSIC') return null;
final t = m['title'] is Map ? m['title'] as Map : const {};
final romaji = t['romaji'] is String ? t['romaji'] as String : '';
final english = t['english'] is String ? t['english'] as String : '';
final cov = m['coverImage'] is Map ? m['coverImage'] as Map : const {};
return Airing(
anilistId: id,
title: english.isNotEmpty ? english : (romaji.isNotEmpty ? romaji : '?'),
titleRomaji: romaji,
episode: ep,
airingAt: DateTime.fromMillisecondsSinceEpoch(at * 1000, isUtc: true),
cover: cov['large'] is String ? cov['large'] as String : null,
banner: m['bannerImage'] is String ? m['bannerImage'] as String : null,
color: parseHexColor(cov['color']),
popularity: m['popularity'] is int ? m['popularity'] as int : 0,
);
}
@override
String toString() => 'Airing($anilistId "$title" ep $episode @ $airingAt)';
}
/// "#e4a15d" -> opaque [Color]; anything else -> null.
Color? parseHexColor(Object? s) {
if (s is! String || s.length != 7 || !s.startsWith('#')) return null;
final v = int.tryParse(s.substring(1), radix: 16);
return v == null ? null : Color(0xff000000 | v);
}
class AniListException implements Exception {
final int? statusCode;
final String message;
const AniListException(this.message, {this.statusCode});
bool get isRateLimited => statusCode == 429;
@override
String toString() => 'AniListException: $message';
}
/// The airings of one range. [staleSince] is set when AniList refused and
/// older cached data (fetched then) is returned instead.
class AniListSchedule {
final List<Airing> airings;
final DateTime fetchedAt;
final DateTime? staleSince;
const AniListSchedule(this.airings, this.fetchedAt, {this.staleSince});
bool get isStale => staleSince != null;
}
class _Cached {
final DateTime fetched;
final int from;
final int to;
final List<Object?> raw;
const _Cached(this.fetched, this.from, this.to, this.raw);
}
class AniListClient {
AniListClient({
String url = aniListDefaultUrl,
HttpClient? http,
Future<Directory?> Function()? cacheDir,
this.memoryTtl = const Duration(minutes: 10),
this.diskTtl = const Duration(hours: 1),
Future<void> Function(Duration)? delay,
DateTime Function()? now,
String? userAgent,
}) : url = Uri.parse(url),
_http = http ?? HttpClient(),
_ownsHttp = http == null,
_cacheDir = cacheDir ?? _defaultCacheDir,
_delay = delay ?? Future<void>.delayed,
_now = now ?? DateTime.now,
userAgent = userAgent ?? 'watch-together/$appVersion';
final Uri url;
final Duration memoryTtl;
final Duration diskTtl;
/// AniList 403s some default agents.
final String userAgent;
final HttpClient _http;
final bool _ownsHttp;
final Future<Directory?> Function() _cacheDir;
final Future<void> Function(Duration) _delay;
final DateTime Function() _now;
/// Pages per range at most (50 airings each).
static const maxPages = 40;
static const perPage = 50;
static const query = r'''
query($from:Int,$to:Int,$page:Int){Page(page:$page,perPage:50){pageInfo{hasNextPage}
airingSchedules(airingAt_greater:$from,airingAt_lesser:$to,sort:TIME){airingAt episode
media{id title{romaji english} coverImage{large color} bannerImage isAdult popularity format}}}}''';
final _mem = <int, _Cached>{};
final _inflight = <String, Future<AniListSchedule>>{};
static Future<Directory?> _defaultCacheDir() async {
try {
return await getApplicationCacheDirectory();
} catch (_) {
return null;
}
}
/// Every airing in [from, to), cached; see the file comment. [force]
/// skips the fresh caches (stale data is still the fallback).
Future<AniListSchedule> schedule(DateTime from, DateTime to, {bool force = false}) {
final f = from.millisecondsSinceEpoch ~/ 1000, t = to.millisecondsSinceEpoch ~/ 1000;
final k = '$f-$t-$force';
return _inflight[k] ??= _schedule(f, t, force).whenComplete(() {
_inflight.remove(k);
});
}
Future<AniListSchedule> _schedule(int from, int to, bool force) async {
final now = _now();
final mem = _mem[from];
final memHit = mem != null && mem.to == to ? mem : null;
if (!force && memHit != null && now.difference(memHit.fetched) < memoryTtl) {
return _result(memHit);
}
final disk = await _readDisk(from, to);
if (!force && disk != null && now.difference(disk.fetched) < diskTtl) {
_mem[from] = disk;
return _result(disk);
}
try {
final raw = await _fetch(from, to);
final c = _mem[from] = _Cached(_now(), from, to, raw);
await _writeDisk(c);
return _result(c);
} catch (e) {
final stale = [memHit, disk].whereType<_Cached>().fold<_Cached?>(
null, (a, b) => a == null || b.fetched.isAfter(a.fetched) ? b : a);
if (stale == null) rethrow;
_log.warning('AniList failed, using cache from ${stale.fetched}: $e');
return _result(stale, stale: true);
}
}
AniListSchedule _result(_Cached c, {bool stale = false}) => AniListSchedule(
[for (final r in c.raw) ?Airing.fromJson(r)],
c.fetched,
staleSince: stale ? c.fetched : null,
);
Future<List<Object?>> _fetch(int from, int to) async {
final out = <Object?>[];
for (var page = 1; page <= maxPages; page++) {
final data = await _page(from, to, page);
final rows = data['airingSchedules'];
if (rows is List) out.addAll(rows);
final info = data['pageInfo'];
if (info is! Map || info['hasNextPage'] != true) break;
}
return out;
}
Future<Map> _page(int from, int to, int page) async {
for (var attempt = 0;; attempt++) {
final (status, body, retryAfter) = await _post(jsonEncode({
'query': query,
'variables': {'from': from - 1, 'to': to, 'page': page},
}));
if (status == 429 && attempt == 0) {
final wait = retryAfter ?? 60;
_log.info('AniList rate limited, retrying in ${wait}s');
await _delay(Duration(seconds: wait.clamp(1, 65)));
continue;
}
Object? j;
try {
j = jsonDecode(body);
} catch (_) {}
if (status != 200) {
throw AniListException('AniList HTTP $status${_errorText(j)}', statusCode: status);
}
final data = j is Map && j['data'] is Map ? (j['data'] as Map)['Page'] : null;
if (data is! Map) {
throw AniListException('AniList: bad response${_errorText(j)}', statusCode: status);
}
return data;
}
}
static String _errorText(Object? j) {
final errs = j is Map ? j['errors'] : null;
if (errs is! List || errs.isEmpty) return '';
final m = errs.first is Map ? (errs.first as Map)['message'] : null;
return m is String ? ': $m' : '';
}
Future<(int, String, int?)> _post(String body) async {
try {
final req = await _http.postUrl(url).timeout(const Duration(seconds: 15));
req.headers
..contentType = ContentType.json
..set(HttpHeaders.acceptHeader, 'application/json')
..set(HttpHeaders.userAgentHeader, userAgent);
req.write(body);
final res = await req.close().timeout(const Duration(seconds: 20));
final text = await res.transform(const Utf8Decoder(allowMalformed: true)).join();
return (
res.statusCode,
text,
int.tryParse(res.headers.value(HttpHeaders.retryAfterHeader) ?? ''),
);
} on TimeoutException {
throw const AniListException('AniList timed out');
} on IOException catch (e) {
throw AniListException('AniList unreachable: $e');
}
}
// ---------------------------------------------------------------- disk
Future<File?> _file(int from) async {
final dir = await _cacheDir();
return dir == null ? null : File('${dir.path}/anilist-schedule-$from.json');
}
Future<_Cached?> _readDisk(int from, int to) async {
try {
final f = await _file(from);
if (f == null || !await f.exists()) return null;
final j = jsonDecode(await f.readAsString());
if (j is! Map || j['from'] != from || j['to'] != to) return null;
final fetched = j['fetched'], raw = j['raw'];
if (fetched is! int || raw is! List) return null;
return _Cached(DateTime.fromMillisecondsSinceEpoch(fetched), from, to, raw);
} catch (_) {
return null;
}
}
Future<void> _writeDisk(_Cached c) async {
try {
final f = await _file(c.from);
if (f == null) return;
await f.parent.create(recursive: true);
await f.writeAsString(jsonEncode({
'fetched': c.fetched.millisecondsSinceEpoch,
'from': c.from,
'to': c.to,
'raw': c.raw,
}));
// Older ranges are useless once a day has passed.
await for (final e in f.parent.list()) {
final name = e.uri.pathSegments.last;
if (e is File &&
e.path != f.path &&
name.startsWith('anilist-schedule-') &&
_now().difference(await e.lastModified()) > const Duration(days: 1)) {
await e.delete();
}
}
} catch (e) {
_log.warning('AniList cache write failed: $e');
}
}
/// Drops the memory cache (the disk cache stays as the stale fallback).
void clearMemory() => _mem.clear();
void close() {
if (_ownsHttp) _http.close(force: true);
}
}
@@ -0,0 +1,686 @@
import 'dart:async';
import 'dart:convert';
import 'dart:io';
import 'package:sync_relay_core/core.dart';
import '../jellyfin/accounts.dart';
import '../update/updater.dart' show appVersion;
// AnimeFlow app API: what the media server tracks, has, downloads and was
// asked for. Authenticated with the user's own Jellyfin access token
// (X-Emby-Token) on the Jellyfin server next to it.
//
// Errors are JSON {"detail","code"} ([AnimeFlowException]); nginx in front
// may answer 429 with HTML (5 req/s, burst 20), retried with backoff.
// /schedule, /series, /series/{id} and /requests send weak ETags: the last
// body is kept and revalidated with If-None-Match. Times are UTC ISO
// ("2026-10-08T19:36:32Z"). Scopes can change at runtime, so /me is cached
// for a minute at most.
final _log = createLogger('animeflow');
const animeFlowDefaultBase = 'https://anime.senkochan.moe/api/app/v1';
// ---------------------------------------------------------------- errors
enum AnimeFlowErrorCode {
missingToken('missing_token'),
badToken('bad_token'),
revoked('revoked'),
missingScope('missing_scope'),
ownAccount('own_account'),
jellyfinUnreachable('jellyfin_unreachable'),
searchUnavailable('search_unavailable'),
invalidRequest('invalid_request'),
invalidTime('invalid_time'),
invalidRange('invalid_range'),
notFound('not_found'),
seriesNotFound('series_not_found'),
episodeNotFound('episode_not_found'),
requestNotFound('request_not_found'),
userNotFound('user_not_found'),
alreadyTracked('already_tracked'),
alreadyAvailable('already_available'),
alreadyRequested('already_requested'),
invalidState('invalid_state'),
/// nginx 429 (client side, no JSON body).
rateLimited('rate_limited'),
/// No answer: DNS, connect, timeout (client side).
network('network'),
/// Anything not in this list.
unknown('unknown');
const AnimeFlowErrorCode(this.wire);
final String wire;
static AnimeFlowErrorCode parse(Object? s) =>
values.where((c) => c.wire == s).firstOrNull ?? unknown;
/// Best guess from the status alone (body was not JSON).
static AnimeFlowErrorCode forStatus(int status) => switch (status) {
401 => badToken,
403 => missingScope,
404 => notFound,
409 => invalidState,
400 => invalidRequest,
429 => rateLimited,
503 => jellyfinUnreachable,
_ => unknown,
};
}
class AnimeFlowException implements Exception {
/// HTTP status; 0 when no answer came.
final int statusCode;
final AnimeFlowErrorCode code;
final String detail;
const AnimeFlowException(this.statusCode, this.code, this.detail);
/// Sign in again (token missing, wrong or expired).
bool get isUnauthorized => statusCode == 401;
/// Revoked, missing scope or own account.
bool get isForbidden => statusCode == 403;
bool get isNotFound => statusCode == 404;
bool get isConflict => statusCode == 409;
bool get isUnavailable => statusCode == 503 || code == AnimeFlowErrorCode.network;
bool get isRateLimited => code == AnimeFlowErrorCode.rateLimited;
@override
String toString() => 'AnimeFlowException($statusCode ${code.wire}): $detail';
}
// ---------------------------------------------------------------- models
DateTime? _time(Object? v) => v is String ? DateTime.tryParse(v)?.toUtc() : null;
int? _int(Object? v) => v is int ? v : (v is num ? v.toInt() : null);
String? _str(Object? v) => v is String ? v : null;
List<Map> _items(Object? j) =>
j is Map && j['items'] is List ? (j['items'] as List).whereType<Map>().toList() : const [];
/// Formats [d] as AnimeFlow wants it: "2026-10-08T19:36:32Z".
String animeFlowTime(DateTime d) => '${d.toUtc().toIso8601String().split('.').first.replaceAll('Z', '')}Z';
enum AnimeFlowScope {
view,
request,
manage;
static Set<AnimeFlowScope> parse(Object? j) => {
if (j is List)
for (final s in j) ...values.where((v) => v.name == s),
};
}
class AnimeFlowMe {
final String userId;
final String name;
final Set<AnimeFlowScope> scopes;
const AnimeFlowMe({required this.userId, required this.name, required this.scopes});
bool can(AnimeFlowScope s) => scopes.contains(s);
static AnimeFlowMe fromJson(Object? j) {
final m = j is Map ? j : const {};
return AnimeFlowMe(
userId: _str(m['user_id']) ?? '',
name: _str(m['name']) ?? '',
scopes: AnimeFlowScope.parse(m['scopes']),
);
}
}
enum EpisodeStatus {
upcoming,
missing,
downloading,
available,
unmonitored,
failed;
/// Unknown values read as upcoming.
static EpisodeStatus parse(Object? s) => values.where((v) => v.name == s).firstOrNull ?? upcoming;
}
class AnimeFlowEpisode {
/// 0 = specials.
final int season;
final int episode;
final int? absolute;
final String? title;
final DateTime? airDate;
/// The AniList entry of this episode's season (one AnimeFlow series may
/// span several AniList entries).
final int? anilistId;
final EpisodeStatus status;
/// 0..1, only while downloading.
final double? progress;
/// Only when failed.
final String? reason;
const AnimeFlowEpisode({
required this.season,
required this.episode,
this.absolute,
this.title,
this.airDate,
this.anilistId,
required this.status,
this.progress,
this.reason,
});
static AnimeFlowEpisode fromJson(Map j) => AnimeFlowEpisode(
season: _int(j['season']) ?? 0,
episode: _int(j['episode']) ?? 0,
absolute: _int(j['absolute']),
title: _str(j['episode_title']),
airDate: _time(j['air_date']),
anilistId: _int(j['anilist_id']),
status: EpisodeStatus.parse(j['status']),
progress: (j['progress'] as num?)?.toDouble(),
reason: _str(j['reason']),
);
}
/// A /schedule row: one episode plus its series.
class AnimeFlowScheduleItem {
final AnimeFlowEpisode episode;
final int seriesId;
final String title;
final int? anidbId;
final int? tvdbId;
final String? jellyfinSeriesId;
final bool followed;
const AnimeFlowScheduleItem({
required this.episode,
required this.seriesId,
required this.title,
this.anidbId,
this.tvdbId,
this.jellyfinSeriesId,
this.followed = false,
});
static AnimeFlowScheduleItem fromJson(Map j) => AnimeFlowScheduleItem(
episode: AnimeFlowEpisode.fromJson(j),
seriesId: _int(j['series_id']) ?? 0,
title: _str(j['title']) ?? '',
anidbId: _int(j['anidb_id']),
tvdbId: _int(j['tvdb_id']),
jellyfinSeriesId: _str(j['jellyfin_series_id']),
followed: j['followed'] == true,
);
}
class AnimeFlowScheduleWindow {
final DateTime? from;
final DateTime? to;
final List<AnimeFlowScheduleItem> items;
const AnimeFlowScheduleWindow({this.from, this.to, required this.items});
}
enum SeriesState {
releasing,
released;
static SeriesState parse(Object? s) => s == 'released' ? released : releasing;
}
class AnimeFlowSeries {
final int seriesId;
final String title;
final int? anidbId;
final int? tvdbId;
/// The series' (first) AniList entry; later seasons may have others, see
/// [AnimeFlowEpisode.anilistId].
final int? anilistId;
final String? jellyfinSeriesId;
final SeriesState state;
final bool monitored;
final bool followed;
final int episodesTotal;
final int episodesAvailable;
final DateTime? nextAirDate;
/// Only from [AnimeFlowClient.seriesDetail].
final List<AnimeFlowEpisode> episodes;
const AnimeFlowSeries({
required this.seriesId,
required this.title,
this.anidbId,
this.tvdbId,
this.anilistId,
this.jellyfinSeriesId,
this.state = SeriesState.releasing,
this.monitored = false,
this.followed = false,
this.episodesTotal = 0,
this.episodesAvailable = 0,
this.nextAirDate,
this.episodes = const [],
});
static AnimeFlowSeries fromJson(Map j) => AnimeFlowSeries(
seriesId: _int(j['series_id']) ?? 0,
title: _str(j['title']) ?? '',
anidbId: _int(j['anidb_id']),
tvdbId: _int(j['tvdb_id']),
anilistId: _int(j['anilist_id']),
jellyfinSeriesId: _str(j['jellyfin_series_id']),
state: SeriesState.parse(j['state']),
monitored: j['monitored'] == true,
followed: j['followed'] == true,
episodesTotal: _int(j['episodes_total']) ?? 0,
episodesAvailable: _int(j['episodes_available']) ?? 0,
nextAirDate: _time(j['next_air_date']),
episodes: [
if (j['episodes'] is List)
for (final e in (j['episodes'] as List).whereType<Map>()) AnimeFlowEpisode.fromJson(e),
],
);
}
class AnimeFlowSearchHit {
final int anidbId;
final String title;
final int? anilistId;
/// Set when AnimeFlow already tracks it.
final int? seriesId;
const AnimeFlowSearchHit({required this.anidbId, required this.title, this.anilistId, this.seriesId});
static AnimeFlowSearchHit fromJson(Map j) => AnimeFlowSearchHit(
anidbId: _int(j['anidb_id']) ?? 0,
title: _str(j['title']) ?? '',
anilistId: _int(j['anilist_id']),
seriesId: _int(j['series_id']),
);
}
enum RequestKind {
series,
season,
episode;
static RequestKind parse(Object? s) => values.where((v) => v.name == s).firstOrNull ?? series;
}
enum RequestStatus {
/// Briefly, before auto-approval or a manager decides.
pending,
approved,
searching,
done,
rejected,
failed;
static RequestStatus parse(Object? s) => values.where((v) => v.name == s).firstOrNull ?? pending;
}
class AnimeFlowRequest {
final int id;
final RequestKind kind;
final RequestStatus status;
final String title;
final int? anidbId;
final int? seriesId;
final int? season;
final int? episode;
final String? reason;
final DateTime? createdAt;
final DateTime? updatedAt;
/// Only with scope=all.
final String? userId;
final String? userName;
const AnimeFlowRequest({
required this.id,
required this.kind,
required this.status,
required this.title,
this.anidbId,
this.seriesId,
this.season,
this.episode,
this.reason,
this.createdAt,
this.updatedAt,
this.userId,
this.userName,
});
static AnimeFlowRequest fromJson(Map j) => AnimeFlowRequest(
id: _int(j['id']) ?? 0,
kind: RequestKind.parse(j['kind']),
status: RequestStatus.parse(j['status']),
title: _str(j['title']) ?? '',
anidbId: _int(j['anidb_id']),
seriesId: _int(j['series_id']),
season: _int(j['season']),
episode: _int(j['episode']),
reason: _str(j['reason']),
createdAt: _time(j['created_at']),
updatedAt: _time(j['updated_at']),
userId: _str(j['user_id']),
userName: _str(j['user_name']),
);
}
class AnimeFlowUser {
/// The Jellyfin user id.
final String userId;
final String name;
final Set<AnimeFlowScope> scopes;
final bool revoked;
final DateTime? lastSeenAt;
const AnimeFlowUser({
required this.userId,
required this.name,
required this.scopes,
this.revoked = false,
this.lastSeenAt,
});
static AnimeFlowUser fromJson(Map j) => AnimeFlowUser(
userId: _str(j['user_id']) ?? '',
name: _str(j['name']) ?? '',
scopes: AnimeFlowScope.parse(j['scopes']),
revoked: j['revoked'] == true,
lastSeenAt: _time(j['last_seen_at']),
);
}
// ---------------------------------------------------------------- client
class _Etagged {
final String etag;
final Object? body;
const _Etagged(this.etag, this.body);
}
class AnimeFlowClient {
/// [token] wins over [tokenProvider]; with neither, every call fails with
/// [AnimeFlowErrorCode.missingToken].
AnimeFlowClient({
String baseUrl = animeFlowDefaultBase,
this._token,
this._tokenProvider,
HttpClient? http,
Future<void> Function(Duration)? delay,
DateTime Function()? now,
this.meTtl = const Duration(seconds: 60),
String? userAgent,
}) : baseUrl = baseUrl.endsWith('/') ? baseUrl.substring(0, baseUrl.length - 1) : baseUrl,
_http = http ?? HttpClient(),
_ownsHttp = http == null,
_delay = delay ?? Future<void>.delayed,
_now = now ?? DateTime.now,
userAgent = userAgent ?? 'watch-together/$appVersion';
/// Authenticated with the linked Jellyfin account on the same domain
/// (see [tokenFor]), looked up on every call so a re-link applies.
factory AnimeFlowClient.fromAccounts(
JellyfinAccounts accounts, {
String baseUrl = animeFlowDefaultBase,
HttpClient? http,
}) =>
AnimeFlowClient(
baseUrl: baseUrl,
http: http,
tokenProvider: () async {
await accounts.ensureLoaded(validate: false);
return tokenFor(accounts.accounts, baseUrl);
},
);
final String baseUrl;
final Duration meTtl;
final String userAgent;
final String? _token;
final Future<String?> Function()? _tokenProvider;
final HttpClient _http;
final bool _ownsHttp;
final Future<void> Function(Duration) _delay;
final DateTime Function() _now;
/// Request bodies above this are refused by the server.
static const maxBodyBytes = 16 * 1024;
/// Retries after an nginx 429 (backoff 1 s, 2 s unless Retry-After).
static const rateLimitRetries = 2;
/// The longest /schedule window.
static const maxScheduleRange = Duration(days: 45);
String? _lastToken;
final _etags = <String, _Etagged>{};
({DateTime at, AnimeFlowMe me})? _me;
/// The token of the linked Jellyfin account whose host shares
/// [animeFlowBase]'s parent domain (anime.example.org -> example.org,
/// jellyfin.example.org, ...).
static String? tokenFor(Iterable<JellyfinAccount> accounts, String animeFlowBase) {
final host = Uri.tryParse(animeFlowBase)?.host.toLowerCase() ?? '';
if (host.isEmpty) return null;
final labels = host.split('.');
final parent = labels.length > 2 ? labels.skip(1).join('.') : host;
bool matches(String url) {
final h = Uri.tryParse(url)?.host.toLowerCase() ?? '';
return h.isNotEmpty && (h == parent || h.endsWith('.$parent'));
}
for (final a in accounts) {
if (a.urls.any(matches)) return a.token;
}
return null;
}
Future<String> _authToken() async {
final t = _token ?? await _tokenProvider?.call();
if (t == null || t.isEmpty) {
throw const AnimeFlowException(
401, AnimeFlowErrorCode.missingToken, 'No linked Jellyfin account for AnimeFlow');
}
if (t != _lastToken) {
// Another user: nothing cached belongs to them.
_lastToken = t;
_etags.clear();
_me = null;
}
return t;
}
Uri _uri(String path, [Map<String, String>? query]) =>
Uri.parse('$baseUrl$path').replace(queryParameters: query);
/// Sends one call; JSON body or null (204). [etag]: revalidate the last
/// body of this URL.
Future<Object?> _call(String method, String path,
{Map<String, String>? query, Object? body, bool etag = false}) async {
final token = await _authToken();
final uri = _uri(path, query);
final key = uri.toString();
List<int>? bytes;
if (body != null) {
bytes = utf8.encode(jsonEncode(body));
if (bytes.length > maxBodyBytes) {
throw const AnimeFlowException(400, AnimeFlowErrorCode.invalidRequest, 'Request body too large');
}
}
for (var attempt = 0;; attempt++) {
final cached = etag ? _etags[key] : null;
final (status, text, retryAfter, tag) = await _send(method, uri, token, bytes, cached?.etag);
if (status == 304 && cached != null) return cached.body;
if (status >= 200 && status < 300) {
final j = text.trim().isEmpty ? null : _decode(text, status);
if (etag && tag != null && tag.isNotEmpty) _etags[key] = _Etagged(tag, j);
return j;
}
final err = _error(status, text);
if (err.isRateLimited && attempt < rateLimitRetries) {
final wait = retryAfter != null ? Duration(seconds: retryAfter.clamp(1, 30)) : Duration(seconds: 1 << attempt);
_log.info('AnimeFlow rate limited on $path, retrying in ${wait.inSeconds}s');
await _delay(wait);
continue;
}
throw err;
}
}
static Object? _decode(String text, int status) {
try {
return jsonDecode(text);
} catch (_) {
throw AnimeFlowException(status, AnimeFlowErrorCode.unknown, 'Response is not JSON');
}
}
static AnimeFlowException _error(int status, String text) {
Object? j;
try {
j = jsonDecode(text);
} catch (_) {}
if (j is Map && j['code'] is String) {
final code = AnimeFlowErrorCode.parse(j['code']);
return AnimeFlowException(status, code == AnimeFlowErrorCode.unknown ? AnimeFlowErrorCode.forStatus(status) : code,
_str(j['detail']) ?? 'HTTP $status');
}
// Not ours: nginx (HTML 429/502), a proxy, ...
return AnimeFlowException(status, AnimeFlowErrorCode.forStatus(status),
status == 429 ? 'Too many requests, try again shortly' : 'HTTP $status');
}
Future<(int, String, int?, String?)> _send(
String method, Uri uri, String token, List<int>? body, String? ifNoneMatch) async {
try {
final req = await _http.openUrl(method, uri).timeout(const Duration(seconds: 15));
req.headers
..set(HttpHeaders.acceptHeader, 'application/json')
..set(HttpHeaders.userAgentHeader, userAgent)
..set('X-Emby-Token', token);
if (ifNoneMatch != null) req.headers.set(HttpHeaders.ifNoneMatchHeader, ifNoneMatch);
if (body != null) {
req.headers.contentType = ContentType.json;
req.contentLength = body.length;
req.add(body);
}
final res = await req.close().timeout(const Duration(seconds: 20));
final text = await res.transform(const Utf8Decoder(allowMalformed: true)).join();
return (
res.statusCode,
text,
int.tryParse(res.headers.value(HttpHeaders.retryAfterHeader) ?? ''),
res.headers.value(HttpHeaders.etagHeader),
);
} on TimeoutException {
throw const AnimeFlowException(0, AnimeFlowErrorCode.network, 'AnimeFlow timed out');
} on IOException catch (e) {
throw AnimeFlowException(0, AnimeFlowErrorCode.network, 'AnimeFlow unreachable: $e');
}
}
// ------------------------------------------------------------ endpoints
/// Who the token belongs to and what they may do. Cached for [meTtl]
/// (scopes change at runtime).
Future<AnimeFlowMe> me({bool force = false}) async {
await _authToken();
final c = _me;
if (!force && c != null && _now().difference(c.at) < meTtl) return c.me;
final me = AnimeFlowMe.fromJson(await _call('GET', '/me'));
_me = (at: _now(), me: me);
return me;
}
/// Episodes airing in [from, to) (at most 45 days) [view].
Future<AnimeFlowScheduleWindow> schedule(DateTime from, DateTime to) async {
if (!from.isBefore(to) || to.difference(from) > maxScheduleRange) {
throw const AnimeFlowException(400, AnimeFlowErrorCode.invalidRange, 'from must be before to, at most 45 days');
}
final j = await _call('GET', '/schedule',
query: {'from': animeFlowTime(from), 'to': animeFlowTime(to)}, etag: true);
final m = j is Map ? j : const {};
return AnimeFlowScheduleWindow(
from: _time(m['from']),
to: _time(m['to']),
items: [for (final e in _items(j)) AnimeFlowScheduleItem.fromJson(e)],
);
}
/// Every tracked series [view].
Future<List<AnimeFlowSeries>> series() async =>
[for (final s in _items(await _call('GET', '/series', etag: true))) AnimeFlowSeries.fromJson(s)];
/// One series with all its episodes [view].
Future<AnimeFlowSeries> seriesDetail(int seriesId) async {
final j = await _call('GET', '/series/$seriesId', etag: true);
return AnimeFlowSeries.fromJson(j is Map ? j : const {});
}
/// Follow (or unfollow) a series [view].
Future<void> setFollowed(int seriesId, bool followed) =>
_call(followed ? 'PUT' : 'DELETE', '/series/$seriesId/follow');
/// Titles that can be requested [request].
Future<List<AnimeFlowSearchHit>> search(String q) async => [
for (final h in _items(await _call('GET', '/search', query: {'q': q}))) AnimeFlowSearchHit.fromJson(h),
];
/// Asks for a whole new series by its AniDB id [request].
Future<AnimeFlowRequest> requestSeries(int anidbId) => _request({'anidb_id': anidbId});
/// Asks for one season of a tracked series [request].
Future<AnimeFlowRequest> requestSeason(int seriesId, int season) =>
_request({'series_id': seriesId, 'season': season});
/// Asks for one episode of a tracked series [request].
Future<AnimeFlowRequest> requestEpisode(int seriesId, int season, int episode) =>
_request({'series_id': seriesId, 'season': season, 'episode': episode});
Future<AnimeFlowRequest> _request(Map<String, Object> body) async {
final j = await _call('POST', '/requests', body: body);
return AnimeFlowRequest.fromJson(j is Map ? j : const {});
}
/// Own requests, newest first [view]; [all] = everyone's, with the user
/// [manage].
Future<List<AnimeFlowRequest>> requests({bool all = false}) async => [
for (final r in _items(await _call('GET', '/requests', query: all ? {'scope': 'all'} : null, etag: true)))
AnimeFlowRequest.fromJson(r),
];
/// [manage]. Returns the updated request when the server sends one.
Future<AnimeFlowRequest?> approve(int requestId) async =>
_maybeRequest(await _call('POST', '/requests/$requestId/approve'));
/// [manage].
Future<AnimeFlowRequest?> reject(int requestId, {String? reason}) async => _maybeRequest(
await _call('POST', '/requests/$requestId/reject', body: {'reason': ?reason}));
static AnimeFlowRequest? _maybeRequest(Object? j) =>
j is Map && j['id'] != null ? AnimeFlowRequest.fromJson(j) : null;
/// [manage].
Future<List<AnimeFlowUser>> users() async =>
[for (final u in _items(await _call('GET', '/users'))) AnimeFlowUser.fromJson(u)];
/// Sets a user's scopes (and revocation) [manage]. Returns the updated
/// user when the server sends one.
Future<AnimeFlowUser?> setScopes(String jellyfinUserId, Set<AnimeFlowScope> scopes, {bool? revoked}) async {
final j = await _call('PUT', '/users/${Uri.encodeComponent(jellyfinUserId)}/scopes', body: {
'scopes': [for (final s in AnimeFlowScope.values) if (scopes.contains(s)) s.name],
'revoked': ?revoked,
});
_me = null; // could be our own
return j is Map && j['user_id'] != null ? AnimeFlowUser.fromJson(j) : null;
}
/// Forgets cached bodies and /me (e.g. on a manual refresh).
void clearCache() {
_etags.clear();
_me = null;
}
void close() {
if (_ownsHttp) _http.close(force: true);
}
}
@@ -0,0 +1,259 @@
import 'dart:async';
import 'anilist.dart';
import 'animeflow.dart';
// The Schedule tab's data: AniList says what airs when, AnimeFlow says
// what the media server has of it. [matchSchedule] merges the two and
// [ScheduleRepository] loads a window of both, degrading to AniList alone
// when AnimeFlow is unreachable or not linked.
enum ScheduleStatus {
available,
downloading,
upcoming,
missing,
failed,
unmonitored,
/// AnimeFlow does not track this series (or is unreachable).
notTracked;
static ScheduleStatus of(EpisodeStatus s) => switch (s) {
EpisodeStatus.available => available,
EpisodeStatus.downloading => downloading,
EpisodeStatus.upcoming => upcoming,
EpisodeStatus.missing => missing,
EpisodeStatus.failed => failed,
EpisodeStatus.unmonitored => unmonitored,
};
}
/// One airing with what AnimeFlow knows about it.
class ScheduleEntry {
final Airing airing;
final ScheduleStatus status;
/// 0..1 while downloading.
final double? progress;
/// Why it failed.
final String? reason;
/// AnimeFlow tracks the series.
final bool tracked;
final bool followed;
/// AnimeFlow's series id (follow, requests).
final int? seriesId;
/// The AnimeFlow season of the matched episode (requests).
final int? season;
/// Play target once available.
final String? jellyfinSeriesId;
const ScheduleEntry({
required this.airing,
required this.status,
this.progress,
this.reason,
this.tracked = false,
this.followed = false,
this.seriesId,
this.season,
this.jellyfinSeriesId,
});
@override
String toString() => 'ScheduleEntry(${airing.anilistId}#${airing.episode} $status)';
}
/// Without AnimeFlow: future airings are upcoming, past ones unknown.
ScheduleEntry untrackedEntry(Airing a, DateTime now) => ScheduleEntry(
airing: a,
status: a.airingAt.isAfter(now) ? ScheduleStatus.upcoming : ScheduleStatus.notTracked,
);
String _day(DateTime d) {
final l = d.toLocal();
return '${l.year}-${l.month}-${l.day}';
}
class _SeriesRef {
final int seriesId;
final String? jellyfinSeriesId;
bool followed;
_SeriesRef(this.seriesId, this.jellyfinSeriesId, this.followed);
}
/// Merges AniList [airings] with AnimeFlow's [schedule] rows and tracked
/// [series]. Per airing: the episode with the same AniList id (that
/// episode's season's entry) and episode number, else absolute number,
/// else the same local air day; unmatched airings of a tracked AniList id
/// are missing once aired, upcoming before; the rest are not tracked.
List<ScheduleEntry> matchSchedule(
List<Airing> airings, {
required List<AnimeFlowScheduleItem> schedule,
List<AnimeFlowSeries> series = const [],
DateTime? now,
}) {
final at = now ?? DateTime.now();
final refs = <int, _SeriesRef>{};
for (final s in series) {
final id = s.anilistId;
if (id != null) refs[id] ??= _SeriesRef(s.seriesId, s.jellyfinSeriesId, s.followed);
}
final byEp = <String, AnimeFlowScheduleItem>{};
final byAbs = <String, AnimeFlowScheduleItem>{};
final byDay = <String, AnimeFlowScheduleItem>{};
for (final it in schedule) {
final e = it.episode, id = e.anilistId;
if (id == null) continue;
final ref = refs[id] ??= _SeriesRef(it.seriesId, it.jellyfinSeriesId, it.followed);
if (it.followed) ref.followed = true;
byEp.putIfAbsent('$id#${e.episode}', () => it);
if (e.absolute != null) byAbs.putIfAbsent('$id#${e.absolute}', () => it);
if (e.airDate != null) byDay.putIfAbsent('$id@${_day(e.airDate!)}', () => it);
}
return [
for (final a in airings)
() {
final id = a.anilistId;
final ref = refs[id];
final it = byEp['$id#${a.episode}'] ?? byAbs['$id#${a.episode}'] ?? byDay['$id@${_day(a.airingAt)}'];
if (it != null) {
final e = it.episode;
return ScheduleEntry(
airing: a,
status: ScheduleStatus.of(e.status),
progress: e.status == EpisodeStatus.downloading ? e.progress : null,
reason: e.status == EpisodeStatus.failed ? e.reason : null,
tracked: true,
followed: it.followed || (ref?.followed ?? false),
seriesId: it.seriesId,
season: e.season,
jellyfinSeriesId: it.jellyfinSeriesId ?? ref?.jellyfinSeriesId,
);
}
if (ref == null) return ScheduleEntry(airing: a, status: ScheduleStatus.notTracked);
return ScheduleEntry(
airing: a,
status: a.airingAt.isAfter(at) ? ScheduleStatus.upcoming : ScheduleStatus.missing,
tracked: true,
followed: ref.followed,
seriesId: ref.seriesId,
jellyfinSeriesId: ref.jellyfinSeriesId,
);
}(),
];
}
/// One loaded window. Errors are display text; [entries] is empty only
/// when AniList failed with nothing cached.
class ScheduleSnapshot {
final DateTime from;
final DateTime to;
/// Sorted by air time.
final List<ScheduleEntry> entries;
/// AniList failed and nothing was cached.
final String? aniListError;
/// AniList failed; [entries] come from the cache fetched then.
final DateTime? aniListStaleSince;
/// AnimeFlow is unreachable, refused or not linked: statuses are
/// AniList-only (upcoming / not tracked).
final String? animeFlowError;
/// The structured AnimeFlow failure behind [animeFlowError].
final AnimeFlowException? animeFlowException;
/// Who we are on AnimeFlow (null when it failed).
final AnimeFlowMe? me;
const ScheduleSnapshot({
required this.from,
required this.to,
required this.entries,
this.aniListError,
this.aniListStaleSince,
this.animeFlowError,
this.animeFlowException,
this.me,
});
bool get hasAnimeFlow => me != null;
}
/// Loads AniList and AnimeFlow for a window and merges them.
class ScheduleRepository {
ScheduleRepository({required this.aniList, this.animeFlow, DateTime Function()? now})
: _now = now ?? DateTime.now;
final AniListClient aniList;
/// Null = not configured (AniList only).
final AnimeFlowClient? animeFlow;
final DateTime Function() _now;
/// Last week's Monday 00:00 local .. 4 weeks later.
static ({DateTime from, DateTime to}) defaultWindow(DateTime now) {
final l = now.toLocal();
final from = DateTime(l.year, l.month, l.day - (l.weekday - 1) - 7);
return (from: from, to: DateTime(from.year, from.month, from.day + 28));
}
Future<ScheduleSnapshot> load({DateTime? from, DateTime? to, bool force = false}) async {
final w = defaultWindow(_now());
final start = from ?? w.from, end = to ?? w.to;
final flowF = _animeFlow(start, end, force);
final AniListSchedule ani;
try {
ani = await aniList.schedule(start, end, force: force);
} catch (e) {
unawaited(flowF.then((_) {}, onError: (_) {}));
return ScheduleSnapshot(from: start, to: end, entries: const [], aniListError: _text(e));
}
final now = _now();
final flow = await flowF;
final airings = [...ani.airings]..sort((a, b) => a.airingAt.compareTo(b.airingAt));
final entries = flow.error == null
? matchSchedule(airings, schedule: flow.schedule, series: flow.series, now: now)
: [for (final a in airings) untrackedEntry(a, now)];
return ScheduleSnapshot(
from: start,
to: end,
entries: entries,
aniListStaleSince: ani.staleSince,
animeFlowError: flow.error == null ? null : _text(flow.error!),
animeFlowException: flow.error is AnimeFlowException ? flow.error as AnimeFlowException : null,
me: flow.me,
);
}
Future<({AnimeFlowMe? me, List<AnimeFlowScheduleItem> schedule, List<AnimeFlowSeries> series, Object? error})>
_animeFlow(DateTime from, DateTime to, bool force) async {
final c = animeFlow;
if (c == null) {
return (me: null, schedule: const <AnimeFlowScheduleItem>[], series: const <AnimeFlowSeries>[],
error: 'AnimeFlow is not configured');
}
try {
if (force) c.clearCache();
final me = await c.me();
final (sched, series) = await (c.schedule(from, to), c.series()).wait;
return (me: me, schedule: sched.items, series: series, error: null);
} on ParallelWaitError catch (e) {
final errs = e.errors as (AsyncError?, AsyncError?);
return (
me: null,
schedule: const <AnimeFlowScheduleItem>[],
series: const <AnimeFlowSeries>[],
error: (errs.$1 ?? errs.$2)!.error,
);
} catch (e) {
return (me: null, schedule: const <AnimeFlowScheduleItem>[], series: const <AnimeFlowSeries>[], error: e);
}
}
static String _text(Object e) => switch (e) {
AnimeFlowException(:final code, :final detail) => switch (code) {
AnimeFlowErrorCode.missingToken => 'Link the Jellyfin server to see AnimeFlow statuses',
AnimeFlowErrorCode.badToken => 'AnimeFlow: sign in to Jellyfin again',
AnimeFlowErrorCode.revoked => 'AnimeFlow: access revoked',
AnimeFlowErrorCode.rateLimited => 'AnimeFlow: too many requests, try again shortly',
_ => 'AnimeFlow: $detail',
},
AniListException(:final message) => message,
_ => '$e',
};
}
@@ -0,0 +1,171 @@
import 'dart:convert';
import 'dart:io';
import 'package:flutter_test/flutter_test.dart';
import 'package:watch_together/src/schedule/anilist.dart';
import 'schedule_fake_server.dart';
Map<String, Object?> airing(int id, int ep, int at,
{bool adult = false, String format = 'TV', String? english = 'Show', String? color = '#e4a15d'}) =>
{
'airingAt': at,
'episode': ep,
'media': {
'id': id,
'title': {'romaji': 'Romaji $id', 'english': english},
'coverImage': {'large': 'https://img/$id.jpg', 'color': color},
'bannerImage': null,
'isAdult': adult,
'popularity': id * 10,
'format': format,
},
};
Map<String, Object?> page(List<Object?> rows, {bool next = false}) => {
'data': {
'Page': {
'pageInfo': {'hasNextPage': next},
'airingSchedules': rows,
},
},
};
void main() {
late FakeServer server;
late Directory cache;
var now = DateTime.utc(2026, 10, 8, 12);
final delays = <Duration>[];
final from = DateTime.utc(2026, 9, 28), to = DateTime.utc(2026, 10, 26);
AniListClient client() => AniListClient(
url: server.url,
cacheDir: () async => cache,
delay: (d) async => delays.add(d),
now: () => now,
userAgent: 'watch-together/test',
);
setUp(() async {
now = DateTime.utc(2026, 10, 8, 12);
delays.clear();
cache = await Directory.systemTemp.createTemp('anilist-test');
server = await FakeServer.start((req, res) async {
final vars = (jsonDecode(req.body) as Map)['variables'] as Map;
final p = vars['page'] as int;
json(res, page([
if (p == 1) ...[airing(1, 5, 1759900000), airing(2, 1, 1759910000, adult: true)],
if (p == 2) ...[airing(3, 2, 1759920000, format: 'MUSIC'), airing(4, 12, 1759930000, english: null)],
], next: p == 1));
});
});
tearDown(() async {
await server.close();
await cache.delete(recursive: true);
});
test('pages, filters adult and music, parses fields', () async {
final c = client();
final s = await c.schedule(from, to);
expect(server.seen, hasLength(2));
final first = jsonDecode(server.seen.first.body) as Map;
expect(first['variables'], {'from': from.millisecondsSinceEpoch ~/ 1000 - 1, 'to': to.millisecondsSinceEpoch ~/ 1000, 'page': 1});
expect(server.seen.first.headers.value(HttpHeaders.userAgentHeader), 'watch-together/test');
expect(s.airings.map((a) => a.anilistId), [1, 4]);
final a = s.airings.first;
expect(a.title, 'Show');
expect(a.titleRomaji, 'Romaji 1');
expect(a.episode, 5);
expect(a.airingAt, DateTime.fromMillisecondsSinceEpoch(1759900000000, isUtc: true));
expect(a.airingAt.isUtc, isTrue);
expect(a.cover, 'https://img/1.jpg');
expect(a.color?.toARGB32(), 0xffe4a15d);
expect(a.popularity, 10);
expect(s.airings.last.title, 'Romaji 4'); // no English title
expect(s.isStale, isFalse);
c.close();
});
test('memory cache, then disk cache across clients, expiry refetches', () async {
final c = client();
await c.schedule(from, to);
await c.schedule(from, to);
expect(server.seen, hasLength(2));
now = now.add(const Duration(minutes: 11)); // memory stale, disk fresh
await c.schedule(from, to);
expect(server.seen, hasLength(2));
final c2 = client(); // restart
await c2.schedule(from, to);
expect(server.seen, hasLength(2));
now = now.add(const Duration(hours: 1));
await c2.schedule(from, to);
expect(server.seen, hasLength(4));
// Another range start is another key.
await c2.schedule(from.add(const Duration(days: 7)), to);
expect(server.seen, hasLength(6));
c.close();
c2.close();
});
test('429 waits Retry-After once, then succeeds', () async {
var n = 0;
final ok = server.handler;
server.handler = (req, res) async {
if (n++ == 0) {
res.statusCode = 429;
res.headers.set(HttpHeaders.retryAfterHeader, '7');
res.write('{"errors":[{"message":"Too Many Requests."}]}');
return;
}
await ok(req, res);
};
final s = await client().schedule(from, to);
expect(delays, [const Duration(seconds: 7)]);
expect(s.airings, hasLength(2));
});
test('429 twice fails', () async {
server.handler = (req, res) async {
res.statusCode = 429;
res.headers.set(HttpHeaders.retryAfterHeader, '2');
};
await expectLater(
client().schedule(from, to),
throwsA(isA<AniListException>().having((e) => e.isRateLimited, 'rate limited', isTrue)));
expect(delays, hasLength(1));
expect(server.seen, hasLength(2));
});
test('failure returns stale cache with its fetch time', () async {
final c = client();
final fetched = now;
await c.schedule(from, to);
now = now.add(const Duration(hours: 2));
server.handler = (req, res) async => json(res, {'errors': [{'message': 'down'}]}, status: 500);
final s = await c.schedule(from, to);
expect(s.isStale, isTrue);
expect(s.staleSince, fetched);
expect(s.airings, hasLength(2));
// A fresh client (restart) falls back to the disk copy.
final s2 = await client().schedule(from, to);
expect(s2.staleSince!.isAtSameMomentAs(fetched), isTrue);
});
test('failure without cache throws', () async {
server.handler = (req, res) async => json(res, {'errors': [{'message': 'Bad'}]}, status: 400);
await expectLater(client().schedule(from, to),
throwsA(isA<AniListException>().having((e) => e.message, 'message', contains('Bad'))));
});
test('unreachable is an AniListException', () async {
await server.close();
await expectLater(client().schedule(from, to), throwsA(isA<AniListException>()));
});
test('parseHexColor', () {
expect(parseHexColor('#000000')?.toARGB32(), 0xff000000);
expect(parseHexColor('e4a15d'), isNull);
expect(parseHexColor(null), isNull);
});
}
@@ -0,0 +1,335 @@
import 'dart:convert';
import 'dart:io';
import 'package:flutter_test/flutter_test.dart';
import 'package:watch_together/src/jellyfin/accounts.dart';
import 'package:watch_together/src/schedule/animeflow.dart';
import 'schedule_fake_server.dart';
const token = 'test-token';
final episodeJson = {
'season': 2,
'episode': 3,
'absolute': 15,
'episode_title': 'Three',
'air_date': '2026-10-08T19:36:32Z',
'anilist_id': 222,
'status': 'downloading',
'progress': 0.5,
};
final seriesJson = {
'series_id': 7,
'title': 'Show',
'anidb_id': 70,
'tvdb_id': 700,
'anilist_id': 111,
'jellyfin_series_id': 'jf7',
'state': 'released',
'monitored': true,
'followed': true,
'episodes_total': 24,
'episodes_available': 20,
'next_air_date': null,
};
final requestJson = {
'id': 9,
'kind': 'season',
'title': 'Show S2',
'series_id': 7,
'season': 2,
'status': 'searching',
'created_at': '2026-10-01T10:00:00Z',
'updated_at': '2026-10-02T10:00:00Z',
};
void main() {
late FakeServer server;
final delays = <Duration>[];
var now = DateTime.utc(2026, 10, 8);
AnimeFlowClient client({String? tok = token}) => AnimeFlowClient(
baseUrl: '${server.url}/api/app/v1',
token: tok,
delay: (d) async => delays.add(d),
now: () => now,
);
/// Routes "METHOD /path" (below the base) to a handler.
Future<void> serve(Map<String, FakeRoute> routes) async {
server.handler = (req, res) async {
final path = req.uri.path.replaceFirst('/api/app/v1', '');
final h = routes['${req.method} $path'];
if (h == null) return json(res, {'detail': 'Not found', 'code': 'not_found'}, status: 404);
await h(req, res);
};
}
setUp(() async {
delays.clear();
now = DateTime.utc(2026, 10, 8);
server = await FakeServer.start((req, res) async {});
});
tearDown(() => server.close());
test('/me: token header, scopes, cached for a minute', () async {
await serve({
'GET /me': (req, res) async =>
json(res, {'user_id': 'u1', 'name': 'Foxyz', 'scopes': ['view', 'request', 'bogus']}),
});
final c = client();
final me = await c.me();
expect(server.seen.single.headers.value('x-emby-token'), token);
expect(server.seen.single.headers.value(HttpHeaders.userAgentHeader), startsWith('watch-together/'));
expect(me.userId, 'u1');
expect(me.name, 'Foxyz');
expect(me.scopes, {AnimeFlowScope.view, AnimeFlowScope.request});
expect(me.can(AnimeFlowScope.manage), isFalse);
await c.me();
expect(server.seen, hasLength(1));
now = now.add(const Duration(seconds: 61));
await c.me();
expect(server.seen, hasLength(2));
});
test('/schedule: query in UTC ISO, items parsed, ETag/304 returns cached body', () async {
var etagHits = 0;
await serve({
'GET /schedule': (req, res) async {
if (req.headers.value(HttpHeaders.ifNoneMatchHeader) == 'W/"s1"') {
etagHits++;
res.statusCode = 304;
return;
}
json(res, {
'from': req.uri.queryParameters['from'],
'to': req.uri.queryParameters['to'],
'items': [
{...episodeJson, 'series_id': 7, 'title': 'Show', 'anidb_id': 70, 'tvdb_id': 700,
'jellyfin_series_id': 'jf7', 'followed': true},
],
}, etag: 'W/"s1"');
},
});
final c = client();
final from = DateTime.utc(2026, 9, 28), to = DateTime.utc(2026, 10, 26);
final w = await c.schedule(from, to);
expect(server.seen.first.uri.queryParameters, {'from': '2026-09-28T00:00:00Z', 'to': '2026-10-26T00:00:00Z'});
expect(w.from, from);
final it = w.items.single;
expect(it.seriesId, 7);
expect(it.title, 'Show');
expect(it.anidbId, 70);
expect(it.tvdbId, 700);
expect(it.jellyfinSeriesId, 'jf7');
expect(it.followed, isTrue);
final e = it.episode;
expect((e.season, e.episode, e.absolute, e.title, e.anilistId), (2, 3, 15, 'Three', 222));
expect(e.airDate, DateTime.utc(2026, 10, 8, 19, 36, 32));
expect(e.status, EpisodeStatus.downloading);
expect(e.progress, 0.5);
final again = await c.schedule(from, to);
expect(etagHits, 1);
expect(again.items.single.episode.anilistId, 222);
});
test('/schedule range checked locally', () async {
final c = client();
final from = DateTime.utc(2026, 10, 1);
await expectLater(c.schedule(from, from),
throwsA(isA<AnimeFlowException>().having((e) => e.code, 'code', AnimeFlowErrorCode.invalidRange)));
await expectLater(c.schedule(from, from.add(const Duration(days: 46))), throwsA(isA<AnimeFlowException>()));
expect(server.seen, isEmpty);
});
test('/series, /series/{id} with episodes, follow/unfollow', () async {
await serve({
'GET /series': (req, res) async => json(res, {'items': [seriesJson]}, etag: 'W/"a"'),
'GET /series/7': (req, res) async => json(res, {
...seriesJson,
'episodes': [
{'season': 0, 'episode': 1, 'status': 'available'},
{'season': 1, 'episode': 1, 'status': 'failed', 'reason': 'No release'},
],
}),
'PUT /series/7/follow': (req, res) async => res.statusCode = 204,
'DELETE /series/7/follow': (req, res) async => res.statusCode = 204,
});
final c = client();
final s = (await c.series()).single;
expect(s.seriesId, 7);
expect(s.anilistId, 111);
expect(s.state, SeriesState.released);
expect(s.monitored && s.followed, isTrue);
expect((s.episodesTotal, s.episodesAvailable, s.nextAirDate), (24, 20, null));
final d = await c.seriesDetail(7);
expect(d.episodes.map((e) => e.season), [0, 1]);
expect(d.episodes.last.status, EpisodeStatus.failed);
expect(d.episodes.last.reason, 'No release');
await c.setFollowed(7, true);
await c.setFollowed(7, false);
expect(server.seen.skip(2).map((r) => '${r.method} ${r.uri.path}'),
['PUT /api/app/v1/series/7/follow', 'DELETE /api/app/v1/series/7/follow']);
});
test('/search and the three request bodies', () async {
await serve({
'GET /search': (req, res) async => json(res, {
'items': [
{'anidb_id': 70, 'title': 'Show', 'anilist_id': 111, 'series_id': 7},
{'anidb_id': 71, 'title': 'Other'},
],
}),
'POST /requests': (req, res) async => json(res, {...requestJson, ...jsonDecode(req.body) as Map}, status: 202),
});
final c = client();
final hits = await c.search('show');
expect(server.seen.single.uri.queryParameters['q'], 'show');
expect(hits.map((h) => (h.anidbId, h.title, h.anilistId, h.seriesId)), [(70, 'Show', 111, 7), (71, 'Other', null, null)]);
final r = await c.requestSeries(70);
expect(r.id, 9);
expect(r.status, RequestStatus.searching);
await c.requestSeason(7, 2);
await c.requestEpisode(7, 2, 3);
expect(server.seen.skip(1).map((r) => jsonDecode(r.body)), [
{'anidb_id': 70},
{'series_id': 7, 'season': 2},
{'series_id': 7, 'season': 2, 'episode': 3},
]);
expect(server.seen.last.headers.contentType?.mimeType, 'application/json');
});
test('/requests own and all, approve, reject', () async {
await serve({
'GET /requests': (req, res) async => json(res, {
'items': [
if (req.uri.queryParameters['scope'] == 'all') {...requestJson, 'user_id': 'u2', 'user_name': 'Bob'}
else requestJson,
],
}),
'POST /requests/9/approve': (req, res) async => json(res, {...requestJson, 'status': 'approved'}),
'POST /requests/9/reject': (req, res) async => res.statusCode = 204,
});
final c = client();
final own = (await c.requests()).single;
expect(own.kind, RequestKind.season);
expect((own.seriesId, own.season, own.episode, own.userName), (7, 2, null, null));
expect(own.createdAt, DateTime.utc(2026, 10, 1, 10));
final all = (await c.requests(all: true)).single;
expect(all.userName, 'Bob');
expect((await c.approve(9))?.status, RequestStatus.approved);
expect(await c.reject(9, reason: 'nope'), isNull);
expect(jsonDecode(server.seen.last.body), {'reason': 'nope'});
await c.reject(9);
expect(jsonDecode(server.seen.last.body), <String, Object?>{});
});
test('/users and scopes', () async {
await serve({
'GET /users': (req, res) async => json(res, {
'items': [
{'user_id': 'u2', 'name': 'Bob', 'scopes': ['view'], 'revoked': true, 'last_seen_at': '2026-10-07T08:00:00Z'},
],
}),
'PUT /users/u2/scopes': (req, res) async => res.statusCode = 204,
});
final c = client();
final u = (await c.users()).single;
expect((u.userId, u.name, u.revoked), ('u2', 'Bob', true));
expect(u.scopes, {AnimeFlowScope.view});
expect(u.lastSeenAt, DateTime.utc(2026, 10, 7, 8));
await c.setScopes('u2', {AnimeFlowScope.request, AnimeFlowScope.view}, revoked: false);
expect(jsonDecode(server.seen.last.body), {'scopes': ['view', 'request'], 'revoked': false});
});
test('error codes map to typed exceptions', () async {
const cases = {
'missing_token': (401, AnimeFlowErrorCode.missingToken),
'bad_token': (401, AnimeFlowErrorCode.badToken),
'revoked': (403, AnimeFlowErrorCode.revoked),
'missing_scope': (403, AnimeFlowErrorCode.missingScope),
'own_account': (403, AnimeFlowErrorCode.ownAccount),
'jellyfin_unreachable': (503, AnimeFlowErrorCode.jellyfinUnreachable),
'search_unavailable': (503, AnimeFlowErrorCode.searchUnavailable),
'invalid_request': (400, AnimeFlowErrorCode.invalidRequest),
'invalid_time': (400, AnimeFlowErrorCode.invalidTime),
'invalid_range': (400, AnimeFlowErrorCode.invalidRange),
'not_found': (404, AnimeFlowErrorCode.notFound),
'series_not_found': (404, AnimeFlowErrorCode.seriesNotFound),
'episode_not_found': (404, AnimeFlowErrorCode.episodeNotFound),
'request_not_found': (404, AnimeFlowErrorCode.requestNotFound),
'user_not_found': (404, AnimeFlowErrorCode.userNotFound),
'already_tracked': (409, AnimeFlowErrorCode.alreadyTracked),
'already_available': (409, AnimeFlowErrorCode.alreadyAvailable),
'already_requested': (409, AnimeFlowErrorCode.alreadyRequested),
'invalid_state': (409, AnimeFlowErrorCode.invalidState),
'something_new': (409, AnimeFlowErrorCode.invalidState), // falls back by status
};
final c = client();
for (final MapEntry(key: wire, value: (status, code)) in cases.entries) {
server.handler = (req, res) async => json(res, {'detail': 'because $wire', 'code': wire}, status: status);
try {
await c.series();
fail('no throw for $wire');
} on AnimeFlowException catch (e) {
expect((e.statusCode, e.code, e.detail), (status, code, 'because $wire'), reason: wire);
}
}
expect(delays, isEmpty);
});
test('nginx HTML 429 backs off, then succeeds or gives up', () async {
var n = 0;
server.handler = (req, res) async {
if (n++ < 2) {
res.statusCode = 429;
res.headers.contentType = ContentType.html;
res.write('<html><body><h1>429 Too Many Requests</h1></body></html>');
return;
}
json(res, {'items': []});
};
final c = client();
expect(await c.series(), isEmpty);
expect(delays, [const Duration(seconds: 1), const Duration(seconds: 2)]);
n = -10;
delays.clear();
await expectLater(
c.series(),
throwsA(isA<AnimeFlowException>()
.having((e) => e.isRateLimited, 'rate limited', isTrue)
.having((e) => e.statusCode, 'status', 429)));
expect(delays, hasLength(AnimeFlowClient.rateLimitRetries));
});
test('no token: missing_token without a request; oversized body refused', () async {
await expectLater(client(tok: null).series(),
throwsA(isA<AnimeFlowException>().having((e) => e.code, 'code', AnimeFlowErrorCode.missingToken)));
await expectLater(client().reject(1, reason: 'x' * 17000),
throwsA(isA<AnimeFlowException>().having((e) => e.code, 'code', AnimeFlowErrorCode.invalidRequest)));
expect(server.seen, isEmpty);
});
test('unreachable is a network error', () async {
await server.close();
await expectLater(client().me(),
throwsA(isA<AnimeFlowException>().having((e) => e.code, 'code', AnimeFlowErrorCode.network)));
});
test('tokenFor picks the account on the same parent domain', () {
JellyfinAccount acc(String id, String url) =>
JellyfinAccount(serverId: id, serverName: id, urls: [url], userId: 'u', userName: 'n', token: 'tok-$id');
final accounts = [
acc('a', 'https://jellyfin.other.org'),
acc('b', 'https://media.senkochan.moe:8920'),
];
expect(AnimeFlowClient.tokenFor(accounts, animeFlowDefaultBase), 'tok-b');
expect(AnimeFlowClient.tokenFor([acc('c', 'https://senkochan.moe')], animeFlowDefaultBase), 'tok-c');
expect(AnimeFlowClient.tokenFor([acc('d', 'https://notsenkochan.moe')], animeFlowDefaultBase), isNull);
expect(AnimeFlowClient.tokenFor(accounts, 'https://anime.example.net/api'), isNull);
});
}
@@ -0,0 +1,51 @@
import 'dart:convert';
import 'dart:io';
/// One request the fake server saw.
class SeenRequest {
final String method;
final Uri uri;
final HttpHeaders headers;
final String body;
SeenRequest(this.method, this.uri, this.headers, this.body);
}
typedef FakeRoute = Future<void> Function(SeenRequest req, HttpResponse res);
/// A localhost HTTP server for the Schedule clients: [handler] answers,
/// [seen] records every request.
class FakeServer {
FakeServer._(this._server) : url = 'http://${_server.address.host}:${_server.port}';
final HttpServer _server;
/// Kept after [close], so clients can try an address nobody answers.
final String url;
final seen = <SeenRequest>[];
late FakeRoute handler;
static Future<FakeServer> start(FakeRoute handler) async {
final s = FakeServer._(await HttpServer.bind(InternetAddress.loopbackIPv4, 0))..handler = handler;
s._server.listen((req) async {
final body = await utf8.decodeStream(req);
final r = SeenRequest(req.method, req.uri, req.headers, body);
s.seen.add(r);
try {
await s.handler(r, req.response);
} catch (e, st) {
// ignore: avoid_print
print('fake server handler failed: $e\n$st');
req.response.statusCode = 599;
}
await req.response.close();
});
return s;
}
Future<void> close() => _server.close(force: true);
}
void json(HttpResponse res, Object? body, {int status = 200, String? etag}) {
res.statusCode = status;
res.headers.contentType = ContentType.json;
if (etag != null) res.headers.set(HttpHeaders.etagHeader, etag);
res.write(jsonEncode(body));
}
+242
View File
@@ -0,0 +1,242 @@
import 'dart:convert';
import 'dart:io';
import 'package:flutter_test/flutter_test.dart';
import 'package:watch_together/src/schedule/anilist.dart';
import 'package:watch_together/src/schedule/animeflow.dart';
import 'package:watch_together/src/schedule/schedule_model.dart';
import 'schedule_fake_server.dart';
// Local times, so the "same local air day" rule is timezone-proof.
final now = DateTime(2026, 10, 8, 12);
Airing airing(int id, int ep, DateTime local) => Airing(
anilistId: id,
title: 'Show $id',
titleRomaji: 'Show $id',
episode: ep,
airingAt: local.toUtc(),
);
AnimeFlowScheduleItem item(int? anilistId, int ep, EpisodeStatus status,
{int season = 1, int? absolute, DateTime? air, int seriesId = 7, bool followed = false,
double? progress, String? reason}) =>
AnimeFlowScheduleItem(
episode: AnimeFlowEpisode(
season: season,
episode: ep,
absolute: absolute,
airDate: air?.toUtc(),
anilistId: anilistId,
status: status,
progress: progress,
reason: reason,
),
seriesId: seriesId,
title: 'Show',
jellyfinSeriesId: 'jf$seriesId',
followed: followed,
);
void main() {
group('matchSchedule', () {
test('by AniList id + episode, per season ids', () {
// One AnimeFlow series (7) spans two AniList entries: S1 = 100, S2 = 200.
final out = matchSchedule(
[airing(100, 3, DateTime(2026, 10, 1, 20)), airing(200, 3, DateTime(2026, 10, 9, 20))],
schedule: [
item(100, 3, EpisodeStatus.available, season: 1),
item(200, 3, EpisodeStatus.downloading, season: 2, progress: 0.4, followed: true),
],
series: [const AnimeFlowSeries(seriesId: 7, title: 'Show', anilistId: 100, jellyfinSeriesId: 'jf7')],
now: now,
);
expect(out.map((e) => e.status), [ScheduleStatus.available, ScheduleStatus.downloading]);
expect(out.map((e) => e.season), [1, 2]);
expect(out[1].progress, 0.4);
expect(out[1].followed, isTrue);
expect(out.every((e) => e.tracked && e.seriesId == 7 && e.jellyfinSeriesId == 'jf7'), isTrue);
});
test('episode numbers do not cross AniList ids', () {
final out = matchSchedule(
[airing(200, 1, DateTime(2026, 10, 2, 20))],
schedule: [item(100, 1, EpisodeStatus.available)],
now: now,
);
expect(out.single.status, ScheduleStatus.notTracked);
expect(out.single.tracked, isFalse);
});
test('absolute number, then same local air day', () {
final out = matchSchedule(
[
airing(300, 1100, DateTime(2026, 10, 5, 9)), // AniList counts absolute
airing(400, 13, DateTime(2026, 10, 6, 23, 30)), // numbering differs
],
schedule: [
item(300, 5, EpisodeStatus.available, season: 22, absolute: 1100),
item(400, 1, EpisodeStatus.failed, season: 2, air: DateTime(2026, 10, 6, 0, 15), reason: 'No release'),
],
now: now,
);
expect(out[0].status, ScheduleStatus.available);
expect(out[0].season, 22);
expect(out[1].status, ScheduleStatus.failed);
expect(out[1].reason, 'No release');
});
test('a different local day does not match', () {
final out = matchSchedule(
[airing(400, 13, DateTime(2026, 10, 7, 0, 30))],
schedule: [item(400, 1, EpisodeStatus.available, air: DateTime(2026, 10, 6, 23, 50))],
now: now,
);
// Tracked (the id is in the schedule) but unmatched and aired.
expect(out.single.status, ScheduleStatus.missing);
expect(out.single.tracked, isTrue);
});
test('tracked but unmatched: missing once aired, upcoming before', () {
final out = matchSchedule(
[airing(500, 2, DateTime(2026, 10, 7, 20)), airing(500, 3, DateTime(2026, 10, 14, 20))],
schedule: const [],
series: [const AnimeFlowSeries(seriesId: 9, title: 'S', anilistId: 500, followed: true)],
now: now,
);
expect(out.map((e) => e.status), [ScheduleStatus.missing, ScheduleStatus.upcoming]);
expect(out.every((e) => e.followed && e.seriesId == 9), isTrue);
});
test('progress only while downloading, reason only when failed', () {
final out = matchSchedule(
[airing(600, 1, DateTime(2026, 10, 1, 20))],
schedule: [item(600, 1, EpisodeStatus.available, progress: 1, reason: 'x')],
now: now,
);
expect(out.single.progress, isNull);
expect(out.single.reason, isNull);
});
});
test('defaultWindow: last week Monday, 28 days', () {
final w = ScheduleRepository.defaultWindow(DateTime(2026, 10, 8, 15)); // Thursday
expect(w.from, DateTime(2026, 9, 28));
expect(w.to, DateTime(2026, 10, 26));
final mon = ScheduleRepository.defaultWindow(DateTime(2026, 10, 5, 1));
expect(mon.from, DateTime(2026, 9, 28));
});
group('ScheduleRepository', () {
late FakeServer server;
late Directory cache;
var aniDown = false;
var flowStatus = 200;
final at = DateTime(2026, 10, 1, 20).toUtc();
setUp(() async {
aniDown = false;
flowStatus = 200;
cache = await Directory.systemTemp.createTemp('schedule-repo-test');
server = await FakeServer.start((req, res) async {
if (req.uri.path == '/anilist') {
if (aniDown) return json(res, {'errors': [{'message': 'down'}]}, status: 500);
return json(res, {
'data': {
'Page': {
'pageInfo': {'hasNextPage': false},
'airingSchedules': [
for (final (id, ep, t) in [(100, 3, at), (900, 1, at.add(const Duration(days: 20))), (901, 1, at)])
{
'airingAt': t.millisecondsSinceEpoch ~/ 1000,
'episode': ep,
'media': {'id': id, 'title': {'romaji': 'R$id'}, 'isAdult': false, 'format': 'TV'},
},
],
},
},
});
}
if (flowStatus != 200) {
return json(res, {'detail': 'Jellyfin is down', 'code': 'jellyfin_unreachable'}, status: flowStatus);
}
switch (req.uri.path) {
case '/flow/me':
return json(res, {'user_id': 'u', 'name': 'Me', 'scopes': ['view']});
case '/flow/schedule':
return json(res, {
'items': [
{'season': 1, 'episode': 3, 'anilist_id': 100, 'status': 'available', 'series_id': 7,
'title': 'Show', 'followed': false},
],
});
case '/flow/series':
return json(res, {'items': []});
}
});
});
tearDown(() async {
await server.close();
await cache.delete(recursive: true);
});
ScheduleRepository repo({bool flow = true}) => ScheduleRepository(
aniList: AniListClient(url: '${server.url}/anilist', cacheDir: () async => cache, now: () => now),
animeFlow: flow ? AnimeFlowClient(baseUrl: '${server.url}/flow', token: 'tok', now: () => now) : null,
now: () => now,
);
test('both ok: merged, sorted', () async {
final s = await repo().load();
expect(s.aniListError, isNull);
expect(s.animeFlowError, isNull);
expect(s.me?.name, 'Me');
expect(s.entries.map((e) => (e.airing.anilistId, e.status)), [
(100, ScheduleStatus.available),
(901, ScheduleStatus.notTracked),
(900, ScheduleStatus.notTracked),
]);
final aniReq = server.seen.firstWhere((r) => r.uri.path == '/anilist');
final vars = (jsonDecode(aniReq.body) as Map)['variables'] as Map;
expect(vars['from'], DateTime(2026, 9, 28).millisecondsSinceEpoch ~/ 1000 - 1);
final sched = server.seen.firstWhere((r) => r.uri.path == '/flow/schedule');
expect(sched.uri.queryParameters['from'], animeFlowTime(DateTime(2026, 9, 28)));
});
test('AnimeFlow failing: AniList-only statuses and an error', () async {
flowStatus = 503;
final s = await repo().load();
expect(s.animeFlowException?.code, AnimeFlowErrorCode.jellyfinUnreachable);
expect(s.animeFlowError, contains('Jellyfin is down'));
expect(s.me, isNull);
expect(s.entries.map((e) => (e.airing.anilistId, e.status, e.tracked)), [
(100, ScheduleStatus.notTracked, false),
(901, ScheduleStatus.notTracked, false),
(900, ScheduleStatus.upcoming, false),
]);
});
test('AnimeFlow not configured', () async {
final s = await repo(flow: false).load();
expect(s.animeFlowError, isNotNull);
expect(s.entries, hasLength(3));
});
test('AniList failing: stale cache, else an error', () async {
final r = repo();
aniDown = true;
final fail = await r.load();
expect(fail.entries, isEmpty);
expect(fail.aniListError, contains('500'));
aniDown = false;
await r.load();
aniDown = true;
final stale = await r.load(force: true);
expect(stale.aniListStaleSince, isNotNull);
expect(stale.entries, hasLength(3));
expect(stale.entries.first.status, ScheduleStatus.available);
});
});
}