//
// Copyright (c) Double Symmetry GmbH
// Commercial use requires a license. See https://rntp.dev/pricing
//

import XCTest
@testable import PlayerCore

final class DownloadCoordinatorTests: XCTestCase {
  private var cacheDir: URL!
  private var cache: AudioCache!
  private var coordinator: DownloadCoordinator!

  override func setUp() {
    super.setUp()
    cacheDir = FileManager.default.temporaryDirectory
      .appendingPathComponent("DownloadCoordinatorTests-\(UUID().uuidString)", isDirectory: true)
    cache = AudioCache(maxSizeBytes: 10 * 1024 * 1024, directory: cacheDir)
    coordinator = DownloadCoordinator(cache: cache)
  }

  override func tearDown() {
    coordinator = nil
    try? FileManager.default.removeItem(at: cacheDir)
    super.tearDown()
  }

  func testDownloadCachesDataAndFinalizesContentLength() throws {
    let testData = Data(repeating: 0xAB, count: 1000)
    let server = try LocalAudioServer(
      audioData: testData,
      options: .init(includeContentLength: true, contentType: "audio/mpeg")
    )
    server.start()
    defer { server.stop() }

    let url = server.url
    let key = cache.cacheKey(for: url)
    let done = expectation(description: "download complete")

    coordinator.download(url: url, headers: nil) { result in
      if case .success = result { done.fulfill() }
    }

    wait(for: [done], timeout: 5.0)

    XCTAssertEqual(cache.cachedBytes(for: key), Int64(testData.count))
    let info = cache.contentInfo(for: key)
    XCTAssertNotNil(info)
    XCTAssertEqual(info?.contentLength, Int64(testData.count))
  }

  func testDownloadWithoutContentLengthFinalizesOnCompletion() throws {
    let testData = Data(repeating: 0xCD, count: 500)
    let server = try LocalAudioServer(
      audioData: testData,
      options: .init(includeContentLength: false, contentType: "audio/aac")
    )
    server.start()
    defer { server.stop() }

    let url = server.url
    let key = cache.cacheKey(for: url)
    let done = expectation(description: "download complete")

    coordinator.download(url: url, headers: nil) { result in
      if case .success = result { done.fulfill() }
    }

    wait(for: [done], timeout: 5.0)

    XCTAssertEqual(cache.cachedBytes(for: key), Int64(testData.count))
    let info = cache.contentInfo(for: key)
    XCTAssertNotNil(info)
    XCTAssertEqual(info?.contentLength, Int64(testData.count),
                   "Content length should be finalized to actual byte count")
  }

  func testDuplicateDownloadReusesActiveTask() throws {
    let testData = Data(repeating: 0xEF, count: 2000)
    let server = try LocalAudioServer(
      audioData: testData,
      options: .init(includeContentLength: true, contentType: "audio/mpeg")
    )
    server.start()
    defer { server.stop() }

    let url = server.url
    let key = cache.cacheKey(for: url)

    let done1 = expectation(description: "first subscriber done")
    let done2 = expectation(description: "second subscriber done")

    coordinator.download(url: url, headers: nil) { _ in done1.fulfill() }
    coordinator.download(url: url, headers: nil) { _ in done2.fulfill() }

    wait(for: [done1, done2], timeout: 5.0)

    // Data should only be cached once (not duplicated)
    XCTAssertEqual(cache.cachedBytes(for: key), Int64(testData.count))
  }

  func testCancellingOneSubscriberDoesNotStopDownload() throws {
    let testData = Data(repeating: 0x11, count: 1000)
    let server = try LocalAudioServer(
      audioData: testData,
      options: .init(includeContentLength: true, contentType: "audio/mpeg")
    )
    server.start()
    defer { server.stop() }

    let url = server.url
    let key = cache.cacheKey(for: url)

    let done = expectation(description: "second subscriber done")

    let token1 = coordinator.download(url: url, headers: nil) { _ in }
    coordinator.download(url: url, headers: nil) { _ in done.fulfill() }
    coordinator.cancelDownload(token: token1)

    wait(for: [done], timeout: 5.0)
    XCTAssertEqual(cache.cachedBytes(for: key), Int64(testData.count))
  }

  func testCancellingAllSubscribersCancelsDownload() throws {
    let testData = Data(repeating: 0x22, count: 1000)
    let server = try LocalAudioServer(
      audioData: testData,
      options: .init(includeContentLength: true, contentType: "audio/mpeg")
    )
    server.start()
    defer { server.stop() }

    let url = server.url

    let token1 = coordinator.download(url: url, headers: nil) { _ in }
    let token2 = coordinator.download(url: url, headers: nil) { _ in }
    coordinator.cancelDownload(token: token1)
    coordinator.cancelDownload(token: token2)

    // Give time for cancellation to propagate
    Thread.sleep(forTimeInterval: 0.5)

    XCTAssertTrue(coordinator.activeDownloadCount == 0)
  }

  func testDownloadStopsAtMaxBytes() throws {
    // Use 200 KB so the slow-network path sends multiple 64 KB chunks,
    // giving the coordinator a chance to cancel mid-stream.
    let totalBytes = 200 * 1024
    let testData = Data(repeating: 0xBB, count: totalBytes)
    let server = try LocalAudioServer(
      audioData: testData,
      options: .init(includeContentLength: true, contentType: "audio/mpeg", simulateSlowNetwork: true)
    )
    server.start()
    defer { server.stop() }

    let url = server.url
    let key = cache.cacheKey(for: url)
    // Stop after ~70 KB — well into the second 64 KB chunk but short of the full 200 KB.
    let maxBytes: Int64 = 70 * 1024
    let done = expectation(description: "download stopped at limit")

    coordinator.download(url: url, headers: nil, cacheKey: key, maxBytes: maxBytes) { result in
      if case .success = result { done.fulfill() }
    }

    wait(for: [done], timeout: 10.0)

    let cached = cache.cachedBytes(for: key)
    // Should be at least maxBytes but not the full file.
    // The coordinator checks after each data chunk, so it may slightly overshoot.
    XCTAssertGreaterThanOrEqual(cached, maxBytes)
    XCTAssertLessThan(cached, Int64(testData.count),
                      "Should not have downloaded the full file")
  }

  func testFullyCachedURLSkipsDownload() throws {
    let testData = Data(repeating: 0x33, count: 100)
    let url = URL(string: "https://example.com/cached.mp3")!
    let key = cache.cacheKey(for: url)

    // Pre-populate cache
    let info = AudioCache.ContentInfo(contentType: "audio/mpeg", contentLength: 100, isByteRangeAccessSupported: false)
    cache.storeContentInfo(info, for: key, url: url)
    cache.appendData(testData, for: key)

    let done = expectation(description: "immediate completion")
    coordinator.download(url: url, headers: nil) { result in
      if case .success = result { done.fulfill() }
    }

    wait(for: [done], timeout: 1.0)
    XCTAssertEqual(coordinator.activeDownloadCount, 0, "No download should have started")
  }

  // MARK: - ICY

  /// Build a raw interleaved ICY body: each segment is `interval` audio bytes, a
  /// length byte, and an optional NUL-padded metadata block.
  private func icyBody(interval: Int, segments: [(audio: Data, meta: String?)]) -> Data {
    var body = Data()
    for seg in segments {
      body.append(seg.audio)
      if let meta = seg.meta {
        let metaBytes = Data(meta.utf8)
        let blocks = (metaBytes.count + 15) / 16
        body.append(UInt8(blocks))
        var padded = metaBytes
        padded.append(Data(count: blocks * 16 - metaBytes.count))
        body.append(padded)
      } else {
        body.append(UInt8(0))
      }
    }
    return body
  }

  private func icyHeaderValue(_ name: String, in headers: [String: String]) -> String? {
    headers.first { $0.key.caseInsensitiveCompare(name) == .orderedSame }?.value
  }

  func testInjectsIcyMetadataHeader() throws {
    let server = try LocalAudioServer(
      audioData: Data(repeating: 0x01, count: 50),
      options: .init(includeContentLength: true, contentType: "audio/mpeg"))
    server.start(); defer { server.stop() }

    let done = expectation(description: "done")
    coordinator.download(url: server.url, headers: nil) { _ in done.fulfill() }
    wait(for: [done], timeout: 5.0)

    XCTAssertEqual(icyHeaderValue("Icy-MetaData", in: server.lastRequestHeaders), "1")
  }

  func testCallerIcyMetadataNotOverridden() throws {
    let server = try LocalAudioServer(
      audioData: Data(repeating: 0x01, count: 50),
      options: .init(includeContentLength: true, contentType: "audio/mpeg"))
    server.start(); defer { server.stop() }

    let done = expectation(description: "done")
    coordinator.download(url: server.url, headers: ["icy-metadata": "0"]) { _ in done.fulfill() }
    wait(for: [done], timeout: 5.0)

    XCTAssertEqual(icyHeaderValue("Icy-MetaData", in: server.lastRequestHeaders), "0")
  }

  func testFiniteICYDeinterleavesCachesCleanAudioAndFiresMetadata() throws {
    let interval = 16
    let a1 = Data(repeating: 0xA1, count: interval)
    let a2 = Data(repeating: 0xA2, count: interval)
    let body = icyBody(interval: interval, segments: [(a1, "StreamTitle='X';"), (a2, nil)])

    let server = try LocalAudioServer(routes: [
      "/icy.mp3": .init(data: body, contentType: "audio/mpeg", includeContentLength: true, icyMetaint: interval)
    ])
    server.start(); defer { server.stop() }

    let url = server.url(forPath: "/icy.mp3")
    let key = cache.cacheKey(for: url)

    let metaReceived = expectation(description: "metadata")
    var received: StreamMetadata?
    coordinator.onICYMetadata = { cacheKey, md in
      if cacheKey == key { received = md; metaReceived.fulfill() }
    }
    let done = expectation(description: "done")
    coordinator.download(url: url, headers: nil, cacheKey: key) { _ in done.fulfill() }

    wait(for: [metaReceived, done], timeout: 5.0)

    XCTAssertEqual(received?.title, "X")
    // Only the clean audio (a1 + a2) is cached — metadata blocks stripped.
    XCTAssertEqual(cache.cachedBytes(for: key), Int64(a1.count + a2.count))
    XCTAssertEqual(cache.readData(for: key, offset: 0, length: 32), a1 + a2)
    let info = cache.contentInfo(for: key)
    XCTAssertEqual(info?.isICY, true)
    XCTAssertEqual(info?.contentLength, 32, "finalized to clean length")
    XCTAssertEqual(cache.icyMetadata(for: key).map(\.audioByteOffset), [16])
    XCTAssertEqual(cache.icyMetadata(for: key).map(\.metadata.title), ["X"])
  }

  func testEndlessICYStreamsCleanAudioAndMetadataWithoutCaching() throws {
    let interval = 16
    let body = icyBody(interval: interval, segments: [(Data(repeating: 0xB1, count: interval), "StreamTitle='Y';")])
    let server = try LocalAudioServer(routes: [
      "/live": .init(data: body, contentType: "audio/mpeg", includeContentLength: false, icyMetaint: interval)
    ])
    server.start(); defer { server.stop() }

    let url = server.url(forPath: "/live")
    let key = cache.cacheKey(for: url)

    let responseReceived = expectation(description: "stream response")
    let audioReceived = expectation(description: "clean stream audio")
    let metadataReceived = expectation(description: "stream metadata")
    let finished = expectation(description: "upstream closed")
    var streamed = Data()
    var metadata: StreamMetadata?

    coordinator.onICYMetadata = { cacheKey, value in
      if cacheKey == key {
        metadata = value
        metadataReceived.fulfill()
      }
    }
    let subscriber = DownloadStreamSubscriber(
      onResponse: { contentType in
        XCTAssertEqual(contentType, "audio/mpeg")
        responseReceived.fulfill()
      },
      onData: { data, acknowledge in
        streamed.append(data)
        acknowledge()
        audioReceived.fulfill()
      }
    )
    coordinator.download(
      url: url,
      headers: nil,
      cacheKey: key,
      streamSubscriber: subscriber
    ) { result in
      if case .success = result { finished.fulfill() }
    }

    wait(for: [responseReceived, audioReceived, metadataReceived, finished], timeout: 5.0)

    XCTAssertEqual(cache.cachedBytes(for: key), 0, "endless ICY must not be cached")
    XCTAssertNil(cache.contentInfo(for: key))
    XCTAssertEqual(streamed, Data(repeating: 0xB1, count: interval))
    XCTAssertEqual(metadata?.title, "Y")
    Thread.sleep(forTimeInterval: 0.2)
    XCTAssertEqual(coordinator.activeDownloadCount, 0, "download entry removed")
  }

  func testExplicitLiveNonICYStreamsWithoutCaching() throws {
    let body = Data(repeating: 0xD1, count: 128)
    let server = try LocalAudioServer(routes: [
      "/live": .init(data: body, contentType: "audio/mpeg", includeContentLength: false)
    ])
    server.start(); defer { server.stop() }

    let url = server.url(forPath: "/live")
    let key = cache.cacheKey(for: url)
    let stale = Data(repeating: 0xEE, count: 32)
    cache.storeContentInfo(
      AudioCache.ContentInfo(
        contentType: "audio/mpeg",
        contentLength: Int64(stale.count),
        isByteRangeAccessSupported: true
      ),
      for: key,
      url: url
    )
    cache.appendData(stale, for: key)
    let responseReceived = expectation(description: "stream response")
    let audioReceived = expectation(description: "raw stream audio")
    let finished = expectation(description: "upstream closed")
    var streamed = Data()
    let subscriber = DownloadStreamSubscriber(
      onResponse: { _ in responseReceived.fulfill() },
      onData: { data, acknowledge in
        streamed.append(data)
        acknowledge()
        audioReceived.fulfill()
      }
    )

    coordinator.download(
      url: url,
      headers: nil,
      cacheKey: key,
      streamOnly: true,
      streamSubscriber: subscriber
    ) { result in
      if case .success = result { finished.fulfill() }
    }

    wait(for: [responseReceived, audioReceived, finished], timeout: 5.0)
    XCTAssertEqual(streamed, body)
    XCTAssertEqual(cache.cachedBytes(for: key), 0)
    XCTAssertNil(cache.contentInfo(for: key))
  }

  func testLiveRelayAppliesBackpressureUntilChunkIsAcknowledged() throws {
    let body = Data(repeating: 0xD2, count: 200 * 1024)
    let server = try LocalAudioServer(
      audioData: body,
      options: .init(
        includeContentLength: true,
        contentType: "audio/mpeg",
        simulateSlowNetwork: true
      )
    )
    server.start(); defer { server.stop() }

    let url = server.url
    let key = cache.cacheKey(for: url)
    let firstChunk = expectation(description: "first chunk")
    let secondChunk = expectation(description: "second chunk")
    let noSecondChunkBeforeAck = expectation(description: "no second chunk before ack")
    noSecondChunkBeforeAck.isInverted = true
    var firstAcknowledgement: (() -> Void)?
    var chunkCount = 0
    var firstWasAcknowledged = false

    let subscriber = DownloadStreamSubscriber(
      onResponse: { _ in },
      onData: { _, acknowledge in
        chunkCount += 1
        if chunkCount == 1 {
          firstAcknowledgement = acknowledge
          firstChunk.fulfill()
        } else if chunkCount == 2 {
          if !firstWasAcknowledged {
            noSecondChunkBeforeAck.fulfill()
          }
          secondChunk.fulfill()
          acknowledge()
        } else {
          acknowledge()
        }
      }
    )
    let token = coordinator.download(
      url: url,
      headers: nil,
      cacheKey: key,
      streamOnly: true,
      streamSubscriber: subscriber
    )
    defer { coordinator.cancelDownload(token: token) }

    wait(for: [firstChunk], timeout: 2.0)
    wait(for: [noSecondChunkBeforeAck], timeout: 0.1)
    firstWasAcknowledged = true
    firstAcknowledgement?()
    wait(for: [secondChunk], timeout: 2.0)
    XCTAssertEqual(cache.cachedBytes(for: key), 0)
  }

  func testInvalidMetaintTreatedAsEndless() throws {
    // icy-metaint: 0 with a Content-Length must NOT de-interleave/cache.
    let server = try LocalAudioServer(routes: [
      "/bad": .init(data: Data(repeating: 0xC1, count: 40), contentType: "audio/mpeg", includeContentLength: true, icyMetaint: 0)
    ])
    server.start(); defer { server.stop() }

    let url = server.url(forPath: "/bad")
    let key = cache.cacheKey(for: url)

    let liveDetected = expectation(description: "live detected")
    coordinator.onDirectFallbackRequired = { cacheKey in
      if cacheKey == key { liveDetected.fulfill() }
    }
    let failed = expectation(description: "failure")
    coordinator.download(url: url, headers: nil, cacheKey: key) { result in
      if case .failure = result { failed.fulfill() }
    }

    wait(for: [liveDetected, failed], timeout: 5.0)
    XCTAssertEqual(cache.cachedBytes(for: key), 0)
  }

  func testResumeOfPartialICYPurgesAndRefetchesFromZero() throws {
    let interval = 16
    let a1 = Data(repeating: 0xA1, count: interval)
    let a2 = Data(repeating: 0xA2, count: interval)
    let body = icyBody(interval: interval, segments: [(a1, "StreamTitle='X';"), (a2, nil)])

    let server = try LocalAudioServer(routes: [
      "/icy.mp3": .init(data: body, contentType: "audio/mpeg", includeContentLength: true, icyMetaint: interval)
    ])
    server.start(); defer { server.stop() }

    let url = server.url(forPath: "/icy.mp3")
    let key = cache.cacheKey(for: url)

    // Pre-seed an incomplete ICY key: marker + partial clean bytes, unknown length.
    cache.storeContentInfo(
      AudioCache.ContentInfo(contentType: "audio/mpeg", contentLength: -1, isByteRangeAccessSupported: false, isICY: true),
      for: key, url: url)
    cache.appendData(Data(repeating: 0x99, count: 10), for: key)

    let done = expectation(description: "done")
    coordinator.download(url: url, headers: nil, cacheKey: key) { _ in done.fulfill() }
    wait(for: [done], timeout: 5.0)

    XCTAssertNil(icyHeaderValue("Range", in: server.lastRequestHeaders), "ICY resume must not send Range")
    // Fresh clean length (32), not 10 + 32 — the stale partial was purged.
    XCTAssertEqual(cache.cachedBytes(for: key), Int64(a1.count + a2.count))
    XCTAssertEqual(cache.readData(for: key, offset: 0, length: 32), a1 + a2)
  }

  func testNonICYPartialThatRespondsWithICYPurgesStalePrefix() throws {
    // A non-ICY partial (isICY nil, e.g. from a pre-ICY build) whose URL now
    // serves ICY on a 200 (Range ignored): download() can't know it's ICY yet, so
    // it sends Range — but the response reveals ICY and the stale prefix must be
    // dropped so clean audio starts at offset 0 (not appended after garbage).
    let interval = 16
    let a1 = Data(repeating: 0xA1, count: interval)
    let a2 = Data(repeating: 0xA2, count: interval)
    let body = icyBody(interval: interval, segments: [(a1, "StreamTitle='X';"), (a2, nil)])

    let server = try LocalAudioServer(routes: [
      "/icy.mp3": .init(data: body, contentType: "audio/mpeg",
                        includeContentLength: true, icyMetaint: interval, ignoreRange: true)
    ])
    server.start(); defer { server.stop() }

    let url = server.url(forPath: "/icy.mp3")
    let key = cache.cacheKey(for: url)
    cache.storeContentInfo(
      AudioCache.ContentInfo(contentType: "audio/mpeg", contentLength: 5000, isByteRangeAccessSupported: true, isICY: nil),
      for: key, url: url)
    cache.appendData(Data(repeating: 0x99, count: 2000), for: key)

    let done = expectation(description: "done")
    coordinator.download(url: url, headers: nil, cacheKey: key) { _ in done.fulfill() }
    wait(for: [done], timeout: 5.0)

    // Range WAS sent (stored info wasn't ICY), but stale prefix was purged.
    XCTAssertEqual(icyHeaderValue("Range", in: server.lastRequestHeaders), "bytes=2000-")
    XCTAssertEqual(cache.cachedBytes(for: key), Int64(a1.count + a2.count))
    XCTAssertEqual(cache.readData(for: key, offset: 0, length: 32), a1 + a2)
    XCTAssertEqual(cache.contentInfo(for: key)?.isICY, true)
    XCTAssertEqual(cache.contentInfo(for: key)?.contentLength, 32)
  }

  func testNonICYPartialResumeHonoredAs206ICYIsNotCached() throws {
    // Origin HONORS Range (206) but advertises icy-metaint — the interleaved body
    // starts mid-stream and can't be framed from 0, so abort + reload direct.
    let interval = 16
    let a1 = Data(repeating: 0xA1, count: interval)
    let body = icyBody(interval: interval, segments: [(a1, "StreamTitle='X';")])

    let server = try LocalAudioServer(routes: [
      "/icy.mp3": .init(data: body, contentType: "audio/mpeg", includeContentLength: true, icyMetaint: interval)
    ])
    server.start(); defer { server.stop() }

    let url = server.url(forPath: "/icy.mp3")
    let key = cache.cacheKey(for: url)
    cache.storeContentInfo(
      AudioCache.ContentInfo(contentType: "audio/mpeg", contentLength: 5000, isByteRangeAccessSupported: true, isICY: nil),
      for: key, url: url)
    cache.appendData(Data(repeating: 0x99, count: 2000), for: key)

    let live = expectation(description: "live detected")
    coordinator.onDirectFallbackRequired = { if $0 == key { live.fulfill() } }
    let failed = expectation(description: "failure")
    coordinator.download(url: url, headers: nil, cacheKey: key) { result in
      if case .failure = result { failed.fulfill() }
    }
    wait(for: [live, failed], timeout: 5.0)

    XCTAssertEqual(cache.cachedBytes(for: key), 0, "un-resumable ICY partial purged")
    XCTAssertNil(cache.contentInfo(for: key))
  }

  // MARK: - Lifetime

  func testCoordinatorDeallocatesAfterInvalidate() throws {
    let server = try LocalAudioServer(
      audioData: Data(repeating: 0x01, count: 500),
      options: .init(includeContentLength: true, contentType: "audio/mpeg"))
    server.start(); defer { server.stop() }

    weak var weakCoord: DownloadCoordinator?
    weak var weakCache: AudioCache?
    autoreleasepool {
      let localCache = AudioCache(
        maxSizeBytes: 1 << 20, directory: cacheDir.appendingPathComponent("inv-\(UUID().uuidString)"))
      let coord = DownloadCoordinator(cache: localCache)
      weakCoord = coord
      weakCache = localCache

      let done = expectation(description: "download done")
      done.assertForOverFulfill = false
      coord.download(url: server.url, headers: nil) { _ in done.fulfill() }  // forces session creation
      wait(for: [done], timeout: 5.0)

      coord.invalidate()  // releases the URLSession's strong hold on the delegate
    }

    // URLSession releases its delegate asynchronously on its own queue after
    // invalidation, so allow a bounded window before asserting.
    let deadline = Date().addingTimeInterval(2.0)
    while weakCoord != nil && Date() < deadline {
      RunLoop.current.run(until: Date().addingTimeInterval(0.05))
    }
    XCTAssertNil(weakCoord, "coordinator must deallocate once the session is invalidated")
    XCTAssertNil(weakCache, "cache held by the coordinator must deallocate too")
  }
}