<?php

declare(strict_types=1);

namespace OCA\NgBackup\Tests\Unit\Backend;

use OCA\NgBackup\Backend\SftpRangeFetcher;
use OCA\NgBackup\Backend\SftpReadStream;
use OCA\NgBackup\Tests\Unit\TempDirTrait;
use PHPUnit\Framework\TestCase;

require_once __DIR__ . '/SftpReadStreamTest.php'; // FakeSftpConnection, FakeSftpStorage

/** SftpRangeFetcher against the fake SFTP server of SftpReadStreamTest, several connections at once. */
class SftpRangeFetcherTest extends TestCase {
	use TempDirTrait;

	/** @var array<string, string> */
	private array $files = [];

	protected function setUp(): void {
		SftpReadStream::reset();
		for ($i = 0; $i < 10; $i++) {
			$this->files["/home/repo/packs/p$i"] = random_bytes(3 * 1048576 + $i * 1000);
		}
	}

	/** @return list<FakeSftpConnection> */
	private function connections(int $n, array $options = []): array {
		$result = [];
		for ($c = 0; $c < $n; $c++) {
			$result[] = new FakeSftpConnection($this->files, ...($options[$c] ?? []));
		}
		return $result;
	}

	/** @return list<object> what StorageBackend passes on: each with its SFTP channel open */
	private static function opened(array $connections): array {
		return array_map(fn (FakeSftpConnection $c): object => SftpReadStream::connection(new FakeSftpStorage($c)), $connections);
	}

	/** @return list<array{0:string, 1:int, 2:int}> */
	private function ranges(): array {
		$ranges = [];
		for ($i = 0; $i < 10; $i++) {
			$ranges[] = ["repo/packs/p$i", 1000 * $i, 2 * 1048576 + 333];
		}
		return $ranges;
	}

	/** @return array<int, string|null> sha1 of the contents by position (a failing comparison of MiBs would not fit in memory) */
	private static function collect(\Generator $streams): array {
		$result = [];
		foreach ($streams as $i => $fh) {
			$result[$i] = $fh === null ? null : sha1(stream_get_contents($fh));
		}
		return $result;
	}

	private function expected(array $ranges): array {
		return array_map(fn (array $r): string => sha1(substr($this->files['/home/' . $r[0]], $r[1], $r[2])), $ranges);
	}

	public function testReadsEveryRangeInOrderOverAllConnections(): void {
		$connections = $this->connections(4);
		$ranges = $this->ranges();
		$spool = $this->tempDir();

		$this->assertSame($this->expected($ranges), self::collect(SftpRangeFetcher::fetch(self::opened($connections), '/home/', $ranges, $spool)));

		foreach ($connections as $c) {
			$this->assertGreaterThan(0, count($c->reads), 'every connection reads');
			$this->assertLessThanOrEqual(SftpRangeFetcher::DEPTH, $c->maxInFlight);
			$this->assertSame([], $c->openHandles);
			$this->assertSame(0, $c->pending());
			$this->assertTrue($c->disconnected, 'its own connections end with it');
		}
		$this->assertSame([], glob("$spool/*"), 'nothing left in the spool');
	}

	public function testKeepsSeveralRequestsInFlightOnEachConnection(): void {
		$this->files['/home/repo/packs/big'] = random_bytes(10 * 1048576);
		$connections = $this->connections(2);
		self::collect(SftpRangeFetcher::fetch(self::opened($connections), '/home/', [['repo/packs/big', 0, 10 * 1048576], ['repo/packs/big', 5, 10 * 1048576 - 5]], $this->tempDir()));

		$this->assertSame(SftpRangeFetcher::DEPTH, $connections[0]->maxInFlight);
		$this->assertSame(SftpRangeFetcher::DEPTH, $connections[1]->maxInFlight);
	}

	public function testAnAnswerPhpseclibHasReadAlreadyIsTakenWithoutWaitingOnTheSocket(): void {
		// The fake's answers are all "read already": its socket never becomes readable, and
		// phpseclib's packet buffer holds data. Waiting on the socket would take a second a step.
		$connections = $this->connections(2);
		$sockets = [];
		foreach ($connections as $c) {
			$pair = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
			$sockets[] = $pair;
			$c->fsock = $pair[0];
			$c->packet_buffer = pack('N', 4) . 'more'; // a whole packet
		}
		$ranges = array_slice($this->ranges(), 0, 3);
		$start = microtime(true);

		$this->assertSame($this->expected($ranges), self::collect(SftpRangeFetcher::fetch(self::opened($connections), '/home/', $ranges, $this->tempDir())));

		$this->assertLessThan(1.0, microtime(true) - $start);
	}

	public function testAServerThatCapsAReadStillGivesEveryByte(): void {
		$connections = $this->connections(3, array_fill(0, 3, ['cap' => 100000]));
		$ranges = $this->ranges();

		$this->assertSame($this->expected($ranges), self::collect(SftpRangeFetcher::fetch(self::opened($connections), '/home/', $ranges, $this->tempDir())));
		// learned: after the first short answers (and asking for their rest) it asks for no more than the server gives
		foreach ($connections as $c) {
			$this->assertLessThanOrEqual(2 * SftpRangeFetcher::DEPTH, count(array_filter($c->reads, fn (int $n): bool => $n > 100000)));
		}
	}

	public function testAMissingFileIsLeftToTheCaller(): void {
		$connections = $this->connections(2);
		$ranges = $this->ranges();
		$ranges[3][0] = 'repo/packs/gone';

		$result = self::collect(SftpRangeFetcher::fetch(self::opened($connections), '/home/', $ranges, $this->tempDir()));

		$expected = $this->expected($this->ranges());
		$expected[3] = null;
		$this->assertSame($expected, $result);
		$this->assertSame([], $connections[0]->openHandles);
	}

	public function testARangePastTheEndOfTheFileIsLeftToTheCaller(): void {
		$ranges = $this->ranges();
		$ranges[5][2] = 10 * 1048576;

		$result = self::collect(SftpRangeFetcher::fetch(self::opened($this->connections(2)), '/home/', $ranges, $this->tempDir()));

		$this->assertNull($result[5]);
		$this->assertSame($this->expected($this->ranges())[6], $result[6]);
	}

	public function testAServerErrorFailsOnlyThatRange(): void {
		// every read from 1 MiB on fails on connection 0: its ranges, not those of the others
		$connections = $this->connections(2, [['failAt' => 1048576]]);
		$ranges = $this->ranges();

		$result = self::collect(SftpRangeFetcher::fetch(self::opened($connections), '/home/', $ranges, $this->tempDir()));

		$expected = $this->expected($ranges);
		$failed = array_keys(array_filter($result, fn (?string $r): bool => $r === null));
		$this->assertNotSame([], $failed);
		foreach ($result as $i => $data) {
			if ($data !== null) {
				$this->assertSame($expected[$i], $data);
			}
		}
		$this->assertLessThan(10, count($failed), 'the other connection goes on');
		$this->assertSame([], $connections[0]->openHandles, 'a failed range is closed too');
	}

	public function testALostConnectionIsNotUsedAgainAndTheOthersGoOn(): void {
		$connections = $this->connections(3, [['loseAt' => 1048576]]);
		$ranges = $this->ranges();

		$result = self::collect(SftpRangeFetcher::fetch(self::opened($connections), '/home/', $ranges, $this->tempDir()));

		$expected = $this->expected($ranges);
		$this->assertSame(1, count(array_filter($result, fn (?string $r): bool => $r === null)), 'only the range it was reading');
		foreach ($result as $i => $data) {
			if ($data !== null) {
				$this->assertSame($expected[$i], $data);
			}
		}
		$this->assertTrue($connections[0]->disconnected);
		$this->assertLessThanOrEqual(1048576 / 261120 + SftpRangeFetcher::DEPTH + 1, count($connections[0]->reads), 'not used after it broke');
	}

	public function testWhenEveryConnectionIsLostTheRestIsLeftToTheCaller(): void {
		$connections = $this->connections(2, [['throwAt' => 1048576], ['loseAt' => 1048576]]);

		$result = self::collect(SftpRangeFetcher::fetch(self::opened($connections), '/home/', $this->ranges(), $this->tempDir()));

		$this->assertSame(array_fill(0, 10, null), $result);
	}

	public function testABrokenConnectionIsReplacedAndTheRestIsReadAhead(): void {
		$connections = $this->connections(1, [['loseAt' => 1048576]]);
		$made = [];
		$connect = function () use (&$made): object {
			$made[] = $c = new FakeSftpConnection($this->files);
			return SftpReadStream::connection(new FakeSftpStorage($c));
		};
		$ranges = $this->ranges();

		$result = self::collect(SftpRangeFetcher::fetch(self::opened($connections), '/home/', $ranges, $this->tempDir(), $connect));

		$expected = $this->expected($ranges);
		$expected[0] = null; // the range it was reading when it broke: read the usual way
		$this->assertSame($expected, $result);
		$this->assertCount(1, $made);
		$this->assertTrue($made[0]->disconnected, 'the new one ends with the fetcher too');
	}

	public function testConnectionsThatKeepBreakingAreReplacedOnlySoOften(): void {
		$made = 0;
		$connect = function () use (&$made): object {
			$made++;
			return SftpReadStream::connection(new FakeSftpStorage(new FakeSftpConnection($this->files, loseAt: 0)));
		};
		$ranges = array_merge($this->ranges(), $this->ranges(), $this->ranges());

		$result = self::collect(SftpRangeFetcher::fetch(self::opened($this->connections(1, [['loseAt' => 0]])), '/home/', $ranges, $this->tempDir(), $connect));

		$this->assertSame(array_fill(0, count($ranges), null), $result);
		$this->assertSame(SftpRangeFetcher::MAX_RECONNECTS, $made);
	}

	public function testStoppingHalfwayEndsTheConnectionsAndLeavesNothingBehind(): void {
		$connections = $this->connections(2);
		$spool = $this->tempDir();
		$streams = SftpRangeFetcher::fetch(self::opened($connections), '/home/', $this->ranges(), $spool);
		$this->assertIsResource($streams->current());

		unset($streams);

		$this->assertTrue($connections[0]->disconnected);
		$this->assertTrue($connections[1]->disconnected);
		$this->assertSame([], glob("$spool/*"));
	}

	public function testReadsNoMoreThanAheadOfTheCaller(): void {
		$connections = $this->connections(4);
		$streams = SftpRangeFetcher::fetch(self::opened($connections), '/home/', $this->ranges(), $this->tempDir());
		$streams->current();

		$opened = array_sum(array_map(fn (FakeSftpConnection $c): int => count(array_unique(array_values($c->opened))), $connections));
		$this->assertLessThanOrEqual(SftpRangeFetcher::AHEAD, $opened);
	}
}