import CoreVideo
import Foundation

/// The app's half: listens for the extension for as long as a call is up,
/// and hands each connection to the call.
///
/// One broadcast at a time is all iOS allows, so one connection at a time
/// is all this takes; a second, which would be a bug, is closed at once.
public final class BroadcastInbox: @unchecked Sendable {
  private let path: String
  private let queue = DispatchQueue(label: "calls.broadcast.inbox")
  private var listener: DispatchSourceRead?
  private var current: BroadcastConnection?

  /// Called on the main queue for each extension that connects.
  public var onConnection: ((BroadcastConnection) -> Void)?

  public init(path: String) { self.path = path }

  public func start() throws {
    let fd = try BroadcastSocket.listen(path: path)
    BroadcastSocket.nonBlocking(fd)
    let source = DispatchSource.makeReadSource(fileDescriptor: fd, queue: queue)
    source.setEventHandler { [weak self] in self?.accept(fd) }
    source.setCancelHandler { Darwin.close(fd) }
    listener = source
    source.resume()
  }

  /// Stops listening. A connection already handed out is the call's to
  /// close, not this.
  public func stop() {
    queue.sync {
      listener?.cancel()
      listener = nil
      current = nil
    }
    unlink(path)
  }

  private func accept(_ fd: Int32) {
    let client = Darwin.accept(fd, nil, nil)
    guard client >= 0 else { return }
    if let current, !current.isClosed {
      Darwin.close(client)
      return
    }
    var on: Int32 = 1
    setsockopt(client, SOL_SOCKET, SO_NOSIGPIPE, &on, socklen_t(MemoryLayout<Int32>.size))
    let connection = BroadcastConnection(fd: client)
    current = connection
    DispatchQueue.main.async { [weak self] in
      guard let self, let onConnection = self.onConnection else {
        connection.close()
        return
      }
      onConnection(connection)
    }
  }
}

/// One broadcast: frames in, until either side closes.
public final class BroadcastConnection: @unchecked Sendable {
  private let fd: Int32
  private let queue = DispatchQueue(label: "calls.broadcast.read", qos: .userInteractive)
  private var source: DispatchSourceRead?
  private let lock = NSLock()
  private var closed = false
  private var header = [UInt8](repeating: 0, count: BroadcastFrameHeader.size)
  private var headerRead = 0
  private var frame: BroadcastFrameHeader?
  private var payload: UnsafeMutableRawPointer?
  private var payloadCapacity = 0
  private var payloadRead = 0
  private var pool: CVPixelBufferPool?
  private var poolShape: (Int, Int, Bool) = (0, 0, false)

  /// Each complete frame, as an NV12 pixel buffer the receiver may keep,
  /// with its CGImagePropertyOrientation. Called on the reading queue.
  public var onFrame: ((CVPixelBuffer, UInt16) -> Void)?
  /// Called once, on the reading queue, when the extension has gone — the
  /// broadcast stopped from Control Center or the status bar, or the
  /// extension ended. Not called for `close()`.
  public var onClose: (() -> Void)?

  init(fd: Int32) { self.fd = fd }

  deinit { payload?.deallocate() }

  var isClosed: Bool {
    lock.lock()
    defer { lock.unlock() }
    return closed
  }

  /// Starts reading. Separate from creation so the reader is only live
  /// once somebody is listening for its frames.
  public func resume() {
    BroadcastSocket.nonBlocking(fd)
    let source = DispatchSource.makeReadSource(fileDescriptor: fd, queue: queue)
    source.setEventHandler { [weak self] in self?.readable() }
    source.setCancelHandler { [fd] in Darwin.close(fd) }
    self.source = source
    source.resume()
  }

  /// Closes from this side. The extension's next write fails, and it ends
  /// the broadcast — which is how "stop sharing" in the app reaches the
  /// system's recording indicator.
  public func close() {
    lock.lock()
    let first = !closed
    closed = true
    lock.unlock()
    guard first else { return }
    if let source {
      source.cancel()
    } else {
      Darwin.close(fd)
    }
  }

  private func ended() {
    lock.lock()
    let first = !closed
    closed = true
    lock.unlock()
    guard first else { return }
    source?.cancel()
    onClose?()
  }

  private func readable() {
    while !isClosed {
      if frame == nil {
        let n = header.withUnsafeMutableBytes {
          Darwin.read(fd, $0.baseAddress!.advanced(by: headerRead), BroadcastFrameHeader.size - headerRead)
        }
        guard let n = settle(n) else { return }
        headerRead += n
        guard headerRead == BroadcastFrameHeader.size else { continue }
        guard let next = header.withUnsafeBytes({ BroadcastFrameHeader.read($0.baseAddress!) }) else {
          ended()
          return
        }
        frame = next
        payloadRead = 0
        if payloadCapacity < next.payloadLength {
          payload?.deallocate()
          payload = UnsafeMutableRawPointer.allocate(byteCount: next.payloadLength, alignment: 16)
          payloadCapacity = next.payloadLength
        }
      }
      guard let frame, let payload else { return }
      let n = Darwin.read(fd, payload.advanced(by: payloadRead), frame.payloadLength - payloadRead)
      guard let n = settle(n) else { return }
      payloadRead += n
      guard payloadRead == frame.payloadLength else { continue }
      self.frame = nil
      headerRead = 0
      if let pixels = picture(frame, payload) { onFrame?(pixels, frame.orientation) }
    }
  }

  /// A read's answer: bytes to count, nil to stop reading for now, and the
  /// connection ended on end-of-file or a real error.
  private func settle(_ n: Int) -> Int? {
    if n > 0 { return n }
    if n < 0 && (errno == EAGAIN || errno == EINTR) { return nil }
    ended()
    return nil
  }

  /// The packed frame, copied into a pixel buffer from a pool, row by row:
  /// the pool's rows are padded and the wire's are not.
  private func picture(_ frame: BroadcastFrameHeader, _ bytes: UnsafeMutableRawPointer)
    -> CVPixelBuffer?
  {
    let shape = (frame.width, frame.height, frame.videoRange)
    if pool == nil || poolShape != shape {
      let format =
        frame.videoRange
        ? kCVPixelFormatType_420YpCbCr8BiPlanarVideoRange
        : kCVPixelFormatType_420YpCbCr8BiPlanarFullRange
      let attributes: [CFString: Any] = [
        kCVPixelBufferPixelFormatTypeKey: format,
        kCVPixelBufferWidthKey: frame.width,
        kCVPixelBufferHeightKey: frame.height,
        kCVPixelBufferIOSurfacePropertiesKey: [:] as [CFString: Any],
      ]
      pool = nil
      CVPixelBufferPoolCreate(nil, nil, attributes as CFDictionary, &pool)
      poolShape = shape
    }
    guard let pool else { return nil }
    var made: CVPixelBuffer?
    CVPixelBufferPoolCreatePixelBuffer(nil, pool, &made)
    guard let pixels = made else { return nil }
    CVPixelBufferLockBaseAddress(pixels, [])
    defer { CVPixelBufferUnlockBaseAddress(pixels, []) }
    let planes = [
      (bytes, frame.height),
      (bytes.advanced(by: frame.width * frame.height), frame.height / 2),
    ]
    for (plane, (source, rows)) in planes.enumerated() {
      guard let target = CVPixelBufferGetBaseAddressOfPlane(pixels, plane) else { return nil }
      let stride = CVPixelBufferGetBytesPerRowOfPlane(pixels, plane)
      for row in 0..<rows {
        target.advanced(by: row * stride).copyMemory(
          from: source.advanced(by: row * frame.width), byteCount: frame.width)
      }
    }
    return pixels
  }
}