import 'dart:async';
import 'dart:convert';
import 'dart:io';

import 'package:flutter_test/flutter_test.dart';
import 'package:jj_mkvmaker/app/settings.dart';
import 'package:jj_mkvmaker/core/ftp.dart';
import 'package:jj_mkvmaker/core/remote_idle.dart';
import 'package:jj_mkvmaker/core/sftp.dart';
import 'package:jj_mkvmaker/core/vfs.dart';
import 'package:jj_mkvmaker/core/webdav.dart';
import 'package:path/path.dart' as p;

import 'support/sftp_server.dart';

/// 관리자 판단 (10/20): 원격 서버의 받기 · 올리기 자료 연결에서 조각 사이에 정한 시간 동안 아무것도 오가지 않으면 실패
/// ("서버가 답하지 않습니다 (n초)"). 예전에는 서버가 멈추면 복사 · 동기화가 영원히 멈춰 있었다.
/// 멈추는 시험 서버 (보내다 멈춤 · 받다 멈춤) 로 정한 시간 뒤 실패하는지 본다 (시험은 1초로).

/// 보내기가 막히도록 큰 자료를 필요한 만큼만 만든다 (서버가 받지 않으면 운영체제 버퍼가 차서 더 가져가지 않음)
Stream<List<int>> _big() async* {
  final chunk = List<int>.filled(64 * 1024, 7);
  for (var i = 0; i < 4096; i++) {
    yield chunk; // 최대 256MB
  }
}

/// 걸린 시간 · 결과 오류
Future<(Duration, Object?)> _timed(Future<void> Function() f) async {
  final sw = Stopwatch()..start();
  Object? err;
  try {
    await f();
  } catch (e) {
    err = e;
  }
  return (sw.elapsed, err);
}

/// FTP 명령만 하는 작은 시험 서버 - RETR 은 조금 보내고 멈추고, STOR 은 자료 연결을 읽지 않는다 (받다 멈춤)
class _StallFtp {
  late ServerSocket _ctl;
  final _open = <Socket>[];
  final _servers = <ServerSocket>[];
  Socket? _data;
  int get port => _ctl.port;

  Future<void> start() async {
    _ctl = await ServerSocket.bind(InternetAddress.loopbackIPv4, 0);
    _ctl.listen((s) {
      _open.add(s);
      void say(String l) => s.write('$l\r\n');
      say('220 stall');
      utf8.decoder.bind(s).transform(const LineSplitter()).listen((line) async {
        switch (line.split(' ').first.toUpperCase()) {
          case 'USER':
            say('331 password');
          case 'PASS':
            say('230 ok');
          case 'FEAT':
            say('211 End');
          case 'PWD':
            say('257 "/" is current');
          case 'EPSV':
            final d = await ServerSocket.bind(InternetAddress.loopbackIPv4, 0);
            _servers.add(d);
            d.listen((x) {
              _open.add(x);
              _data = x; // 읽지 않는다 (STOR 은 받다 멈춤)
            });
            say('229 Entering Extended Passive Mode (|||${d.port}|)');
          case 'RETR':
            say('150 sending');
            for (var i = 0; i < 50 && _data == null; i++) {
              await Future<void>.delayed(const Duration(milliseconds: 20));
            }
            _data?.add(List<int>.filled(1000, 65)); // 조금 보내고 멈춤 (닫지 않음)
          case 'STOR':
            say('150 receiving');
          case 'QUIT':
            say('221 bye');
          default:
            say('200 ok');
        }
      }, onError: (_) {});
    });
  }

  Future<void> stop() async {
    for (final s in _open) {
      s.destroy();
    }
    for (final s in _servers) {
      await s.close();
    }
    await _ctl.close();
  }
}

/// SFTP 시험 서버 앞의 중계: [stall] 이면 양쪽으로 아무것도 넘기지 않는다 (서버가 멈춘 것처럼)
class _StallProxy {
  final int upstream;
  _StallProxy(this.upstream);
  late ServerSocket _ss;
  final _open = <Socket>[];
  bool stall = false;

  Future<void> start() async {
    _ss = await ServerSocket.bind(InternetAddress.loopbackIPv4, 0);
    _ss.listen((c) async {
      final u = await Socket.connect(InternetAddress.loopbackIPv4, upstream);
      _open.addAll([c, u]);
      u.listen((d) {
        if (!stall) c.add(d);
      }, onError: (_) {}, onDone: c.destroy);
      c.listen((d) {
        if (!stall) u.add(d);
      }, onError: (_) {}, onDone: u.destroy);
    });
  }

  Future<Socket> connect(String host, int port, Duration timeout) => Socket.connect(InternetAddress.loopbackIPv4, _ss.port, timeout: timeout);

  Future<void> stop() async {
    for (final s in _open) {
      s.destroy();
    }
    await _ss.close();
  }
}

void main() {
  setUp(() => RemoteIdle.limit = const Duration(seconds: 1));
  tearDown(() => RemoteIdle.limit = const Duration(seconds: RemoteIdle.defaultSeconds));

  test('설정 "원격 서버 응답 기다리기": 처음 60초 · 30 · 60 · 120 · 300 만 · 그 밖의 값은 60 · 오류 글은 초까지', () {
    expect(AppSettings().remoteIdleSeconds, 60);
    for (final n in RemoteIdle.choices) {
      expect(AppSettings.fromJson({'remoteIdleSeconds': n}).remoteIdleSeconds, n);
    }
    for (final v in [0, 45, -1, 'x', null, 99999]) {
      expect(AppSettings.fromJson({'remoteIdleSeconds': v}).remoteIdleSeconds, 60, reason: '$v');
    }
    expect((AppSettings()..remoteIdleSeconds = 120).toJson()['remoteIdleSeconds'], 120);
    expect('${const RemoteIdleTimeout(60)}', '서버가 답하지 않습니다 (60초)');
  });

  test('도우미: 받기 - 조각 사이가 넘으면 실패 · 끊기 · 조각이 계속 오면 실패하지 않음', () async {
    final c = StreamController<List<int>>();
    var cut = false;
    c.add([1]);
    final (t, e) = await _timed(() => idleRead(c.stream, onTimeout: () => cut = true).drain<void>());
    expect(e, isA<RemoteIdleTimeout>());
    expect(t, lessThan(const Duration(seconds: 5)));
    expect(cut, isTrue);
    await c.close();
    // 0.1초마다 조각 (정한 3초 안 - 전체 시험의 부하에서도 넉넉하게, 0b 지적 10/20: 0.4초 대 1초는 한 번 넘음) - 끝까지
    Stream<List<int>> slow() async* {
      for (var i = 0; i < 5; i++) {
        await Future<void>.delayed(const Duration(milliseconds: 100));
        yield [i];
      }
    }

    expect(await idleRead(slow(), limit: const Duration(seconds: 3)).length, 5);
  });

  test('도우미: 올리기 - 보내는 쪽이 더 가져가지 않으면 · 다 가져간 뒤 답이 없으면 실패', () async {
    var cut = false;
    final (t, e) = await _timed(() => idleWrite(_big(), (s) async {
          await s.first; // 한 조각만 가져가고 멈춤
          await Completer<void>().future;
        }, onTimeout: () => cut = true));
    expect(e, isA<RemoteIdleTimeout>());
    expect(t, lessThan(const Duration(seconds: 5)));
    expect(cut, isTrue);
    final (_, e2) = await _timed(() => idleWrite(Stream.value([1]), (s) async {
          await s.drain<void>();
          await Completer<void>().future; // 마지막 답이 없음
        }));
    expect(e2, isA<RemoteIdleTimeout>());
    expect(await idleWrite(Stream.value([1, 2]), (s) => s.expand((x) => x).toList()), [1, 2], reason: '멈추지 않으면 그대로');
  });

  group('FTP (멈추는 시험 서버)', () {
    late _StallFtp server;
    setUp(() async {
      server = _StallFtp();
      await server.start();
      FtpRegistry.configure([FtpServer(id: 'st', name: 'stall', host: '127.0.0.1', port: server.port, user: 'jj', password: 'pw', security: 'none')]);
    });
    tearDown(() async {
      FtpRegistry.configure(const []);
      await server.stop();
    });

    test('받다가 서버가 보내지 않으면 (보내다 멈춤) 정한 시간 뒤 "서버가 답하지 않습니다" · 로컬에 받다 만 파일 없음', () async {
      final tmp = Directory.systemTemp.createTempSync('jj_idleftp_');
      addTearDown(() => tmp.deleteSync(recursive: true));
      final local = p.join(tmp.path, 'a.bin');
      final (t, e) = await _timed(() async => vWrite(local, await vOpenRead('ftp://st/a.bin')));
      expect(e, isA<RemoteIdleTimeout>(), reason: '$e');
      expect('$e', '서버가 답하지 않습니다 (1초)');
      expect(t, lessThan(const Duration(seconds: 10)));
      expect(tmp.listSync(), isEmpty, reason: '받다 만 .jjsync 를 남기지 않음');
    });

    test('올리다가 서버가 받지 않으면 (받다 멈춤) 정한 시간 뒤 실패', () async {
      final (t, e) = await _timed(() => FtpRegistry.client('st').write('/b.bin', _big()));
      expect(e, isA<RemoteIdleTimeout>(), reason: '$e');
      expect(t, lessThan(const Duration(seconds: 15)));
    });
  });

  group('WebDAV (멈추는 시험 서버)', () {
    late HttpServer http;
    final seen = <String>[];
    setUp(() async {
      seen.clear();
      http = await HttpServer.bind(InternetAddress.loopbackIPv4, 0);
      http.listen((req) async {
        seen.add('${req.method} ${req.uri.path}');
        final res = req.response;
        switch ((req.method, req.uri.path)) {
          case ('GET', '/dav/a.bin'):
            res.contentLength = 1 << 20;
            res.add(List<int>.filled(1000, 65)); // 조금 보내고 멈춤
            await res.flush();
          case ('GET', '/dav/nohead.bin'):
            break; // 답 (머리) 을 보내지 않음
          case ('PUT', _):
            break; // 몸을 읽지 않고 답도 없음 (받다 멈춤)
          case ('PROPFIND', _):
            res.statusCode = 404; // 지우기 전 확인 (stat) - 없음으로
            await res.close();
          default:
            res.statusCode = 204;
            await res.close();
        }
      });
      DavRegistry.configure([DavServer(id: 'st', name: 'stall', url: 'http://127.0.0.1:${http.port}/dav')]);
    });
    tearDown(() async {
      DavRegistry.configure(const []);
      await http.close(force: true);
    });

    test('받다가 멈춤 · 답 머리가 없음 → 정한 시간 뒤 실패', () async {
      final (t, e) = await _timed(() async => (await vOpenRead('dav://st/a.bin')).drain<void>());
      expect(e, isA<RemoteIdleTimeout>(), reason: '$e');
      expect(t, lessThan(const Duration(seconds: 10)));
      final (t2, e2) = await _timed(() async => (await vOpenRead('dav://st/nohead.bin')).drain<void>());
      expect(e2, isA<RemoteIdleTimeout>(), reason: '$e2');
      expect(t2, lessThan(const Duration(seconds: 10)));
    });

    test('올리다가 서버가 받지 않으면 실패 · 받다 만 .jjpart 지우기를 서버에 부탁 (그것도 정한 시간까지만)', () async {
      final (t, e) = await _timed(() => vWrite('dav://st/b.bin', _big()));
      expect(e, isA<RemoteIdleTimeout>(), reason: '$e');
      expect(t, lessThan(const Duration(seconds: 20)));
      expect(seen, contains('DELETE /dav/b.bin.jjpart'));
    });
  });

  group('SFTP (paramiko 시험 서버 + 멈추는 중계)', () {
    final has = sftpTestServerAvailable;
    late Directory tmp;
    SftpTestServer? server;
    _StallProxy? proxy;
    setUp(() async {
      if (!has) return;
      tmp = Directory.systemTemp.createTempSync('jj_idlesftp_');
      final root = Directory(p.join(tmp.path, 'root'))..createSync();
      File(p.join(root.path, 'a.bin')).writeAsBytesSync(List<int>.filled(4 << 20, 1));
      server = await SftpTestServer.start(root: root.path, hostKeyFile: p.join(tmp.path, 'hostkey'));
      proxy = _StallProxy(server!.port);
      await proxy!.start();
    });
    tearDown(() async {
      if (!has) return;
      await proxy?.stop();
      await server?.stop();
      for (var i = 0;; i++) {
        try {
          if (tmp.existsSync()) tmp.deleteSync(recursive: true);
          break;
        } on FileSystemException {
          if (i > 30) rethrow;
          await Future<void>.delayed(const Duration(milliseconds: 100));
        }
      }
    });
    SftpRemote client() => SftpRemote(
        SftpServer(id: 's1', name: 'test', host: '127.0.0.1', port: server!.port, user: 'jj', password: 'pw', hostKeyFingerprint: server!.fingerprint),
        connect: proxy!.connect);

    test('받는 중에 서버가 멈추면 · 명령 (열기) 에 답이 없으면 정한 시간 뒤 실패', () async {
      final c = client();
      addTearDown(c.close);
      // 준비 (연결 · 목록 · 열기) 는 넉넉하게 - 함께 도는 시험의 부하에서 paramiko 의 답이 1초를 넘은 적이 있다 (15 검증 10/20)
      RemoteIdle.limit = const Duration(seconds: 10);
      expect((await c.list('/')).map((e) => e.rel), contains('/a.bin'));
      final s = await c.openReadFrom('a.bin'); // 열기까지는 됨
      RemoteIdle.limit = const Duration(seconds: 1); // 받기 (listen) 때 이 시간으로
      proxy!.stall = true; // 이제 서버가 아무것도 보내지 않음
      final (t, e) = await _timed(() => s.drain<void>());
      expect(e, isA<RemoteIdleTimeout>(), reason: '$e');
      expect(t, lessThan(const Duration(seconds: 10)));
      final c2 = client();
      addTearDown(c2.close);
      proxy!.stall = false;
      RemoteIdle.limit = const Duration(seconds: 10);
      await c2.list('/');
      RemoteIdle.limit = const Duration(seconds: 1);
      proxy!.stall = true;
      final (t2, e2) = await _timed(() => c2.openReadFrom('a.bin'));
      expect(e2, isA<RemoteIdleTimeout>(), reason: '$e2');
      expect(t2, lessThan(const Duration(seconds: 10)));
    }, skip: has ? false : 'SFTP 시험 서버 (paramiko) 없음');

    test('올리는 중에 서버가 멈추면 정한 시간 뒤 실패', () async {
      final c = client();
      addTearDown(c.close);
      RemoteIdle.limit = const Duration(seconds: 10); // 준비는 넉넉하게 (위와 같음)
      await c.list('/');
      RemoteIdle.limit = const Duration(seconds: 1);
      proxy!.stall = true;
      final (t, e) = await _timed(() => c.write('b.bin', _big()));
      expect(e, isA<RemoteIdleTimeout>(), reason: '$e');
      expect(t, lessThan(const Duration(seconds: 10)));
    }, skip: has ? false : 'SFTP 시험 서버 (paramiko) 없음');
  });
}