import { execFileSync } from 'node:child_process'
import { PassThrough } from 'node:stream'
import { Buffer } from 'node:buffer'
import { afterEach, beforeEach, describe, expect, it } from 'vitest'
import { RemoteRejectedError, WireError } from '../../server/utils/git-wire/errors'
import { ReceivePackSession, type ReceivePackProcess } from '../../server/utils/git-wire/receive-pack'
import { encodePktLine, flushPkt } from '../../server/utils/git-wire/pkt-line'
import { ZERO_SHA } from '../../server/utils/git-wire/refs'
import { fakeGithubFetch, GitFixture, localReceivePackFactory } from '../utils/git-wire'
import { fetchPack } from '../../server/utils/git-wire/upload-pack'

async function* fromBuffer(b: Buffer): AsyncGenerator<Buffer> {
  yield b
}

async function push(factory: ReturnType<typeof localReceivePackFactory>, updates: Parameters<ReceivePackSession['push']>[0], pack: AsyncIterable<Buffer> | null) {
  const session = await ReceivePackSession.open(factory)
  await session.push(updates, pack)
}

async function drain(gen: AsyncGenerator<Buffer>): Promise<Buffer> {
  const parts: Buffer[] = []
  for await (const c of gen) parts.push(c)
  return Buffer.concat(parts)
}

describe('receive-pack (against real git-receive-pack)', () => {
  let fx: GitFixture
  let realFetch: typeof globalThis.fetch

  beforeEach(() => {
    fx = new GitFixture()
    realFetch = globalThis.fetch
  })

  afterEach(() => {
    globalThis.fetch = realFetch
    fx.cleanup()
  })

  /** Build a pack on disk for `want` and return it as a single buffer. */
  async function packFor(ghBare: string, want: string, haves: string[]): Promise<Buffer> {
    globalThis.fetch = fakeGithubFetch(new Map([['owner/repo', ghBare]])) as unknown as typeof globalThis.fetch
    const { pack } = await fetchPack({ repoFullName: 'owner/repo', token: 't', want, haves, maxBytes: 1 << 30 })
    return drain(pack)
  }

  it('pushes a new ref into an empty knot repo', async () => {
    const gh = fx.initBare('gh.git')
    const work = fx.initWork('work')
    const sha = fx.commit(work, 'a.txt', 'hello')
    fx.pushTo(work, gh, 'HEAD:refs/heads/main')
    const knot = fx.initBare('knot.git')

    const pack = await packFor(gh, sha, [])
    await push(
      localReceivePackFactory(knot),
      [{ ref: 'refs/heads/main', old: ZERO_SHA, next: sha }],
      fromBuffer(pack),
    )

    expect(fx.revParse(knot, 'refs/heads/main')).toBe(sha)
  })

  it('fast-forwards an existing ref with a thin incremental pack', async () => {
    const gh = fx.initBare('gh.git')
    const work = fx.initWork('work')
    const first = fx.commit(work, 'a.txt', 'one')
    fx.pushTo(work, gh, 'HEAD:refs/heads/main')
    const knot = fx.initBare('knot.git')
    await push(localReceivePackFactory(knot), [{ ref: 'refs/heads/main', old: ZERO_SHA, next: first }], fromBuffer(await packFor(gh, first, [])))

    const second = fx.commit(work, 'b.txt', 'two')
    fx.pushTo(work, gh, 'HEAD:refs/heads/main')
    await push(localReceivePackFactory(knot), [{ ref: 'refs/heads/main', old: first, next: second }], fromBuffer(await packFor(gh, second, [first])))

    expect(fx.revParse(knot, 'refs/heads/main')).toBe(second)
  })

  it('rejects a stale compare-and-swap as stale-old-sha', async () => {
    const gh = fx.initBare('gh.git')
    const work = fx.initWork('work')
    const first = fx.commit(work, 'a.txt', 'one')
    fx.pushTo(work, gh, 'HEAD:refs/heads/main')
    const knot = fx.initBare('knot.git')
    await push(localReceivePackFactory(knot), [{ ref: 'refs/heads/main', old: ZERO_SHA, next: first }], fromBuffer(await packFor(gh, first, [])))

    const second = fx.commit(work, 'b.txt', 'two')
    fx.pushTo(work, gh, 'HEAD:refs/heads/main')

    // Claim the knot is still empty when it actually points at `first`.
    await expect(
      push(localReceivePackFactory(knot), [{ ref: 'refs/heads/main', old: ZERO_SHA, next: second }], fromBuffer(await packFor(gh, second, [first]))),
    ).rejects.toMatchObject({ constructor: RemoteRejectedError, reason: 'stale-old-sha' })
  })

  it('deletes a ref with no pack', async () => {
    const gh = fx.initBare('gh.git')
    const work = fx.initWork('work')
    const sha = fx.commit(work, 'a.txt', 'hello')
    fx.pushTo(work, gh, 'HEAD:refs/heads/main')
    const knot = fx.initBare('knot.git')
    await push(localReceivePackFactory(knot), [{ ref: 'refs/heads/main', old: ZERO_SHA, next: sha }], fromBuffer(await packFor(gh, sha, [])))

    await push(localReceivePackFactory(knot), [{ ref: 'refs/heads/main', old: sha, next: ZERO_SHA }], null)

    expect(() => execFileSync('git', ['rev-parse', 'refs/heads/main'], { cwd: knot })).toThrow(/unknown revision|ambiguous argument|fatal/)
  })

  it('kills a stalled session once the watchdog fires', async () => {
    // A factory whose child accepts the connection but never advertises: the
    // open() read would block forever without the watchdog.
    const stalled = () => {
      let resolveDone: (code: number | null) => void
      const done = new Promise<number | null>(r => { resolveDone = r })
      // eslint-disable-next-line require-yield -- models a stalled stream that blocks until killed and never emits
      async function* neverYields(): AsyncGenerator<Buffer> {
        await done
      }
      return {
        stdin: { write: (_d: unknown, cb?: (e?: Error) => void) => cb?.(), end: () => {} } as unknown as NodeJS.WritableStream,
        stdout: neverYields(),
        stderr: () => '',
        kill: () => resolveDone(null),
        done,
      }
    }
    await expect(ReceivePackSession.open(stalled, 50)).rejects.toThrow(/end of stream|advertisement/)
  })

  it('surfaces a mid-push channel death as a transient WireError, not a report parse', async () => {
    // Advertise one ref so open() succeeds, then model the transport dying
    // while the pack streams: stdin closes early and the process exits non-zero
    // with no report-status. The knot in production would report a truncated
    // pack as an `unpack` failure; the session must instead throw a plain
    // (transient, retryable) WireError.
    const factory = () => {
      const stdin = new PassThrough()
      const advertisement = Buffer.concat([
        encodePktLine(`${ZERO_SHA} refs/heads/main\0report-status\n`),
        flushPkt,
      ])
      const stdout = new PassThrough()
      stdout.write(advertisement)
      let resolveDone: (code: number | null) => void
      const done = new Promise<number | null>(r => { resolveDone = r })
      // Let stdin drain (so writeAll + pipePack complete their writes), but end
      // stdout without a report-status flush and exit non-zero, as a reset ssh
      // channel would after eating a partial pack.
      stdin.resume()
      stdin.on('finish', () => {
        stdout.end()
        resolveDone(128)
      })
      const proc: ReceivePackProcess = {
        stdin,
        stdout,
        stderr: () => '',
        kill: () => { stdout.end(); resolveDone(128) },
        done,
      }
      return proc
    }

    async function* pack(): AsyncGenerator<Buffer> {
      yield Buffer.from('PACK')
      yield Buffer.alloc(1024)
    }

    const session = await ReceivePackSession.open(factory)
    await expect(session.push([{ ref: 'refs/heads/main', old: ZERO_SHA, next: 'a'.repeat(40) }], pack()))
      .rejects.toBeInstanceOf(WireError)
  })

  it('pushes an annotated tag', async () => {
    const gh = fx.initBare('gh.git')
    const work = fx.initWork('work')
    fx.commit(work, 'a.txt', 'hello')
    fx.pushTo(work, gh, 'HEAD:refs/heads/main')
    fx.git(['tag', '-a', 'v1', '-m', 'release'], work)
    const tagSha = fx.git(['rev-parse', 'refs/tags/v1'], work)
    fx.pushTo(work, gh, 'refs/tags/v1:refs/tags/v1')
    const knot = fx.initBare('knot.git')

    await push(
      localReceivePackFactory(knot),
      [{ ref: 'refs/tags/v1', old: ZERO_SHA, next: tagSha }],
      fromBuffer(await packFor(gh, tagSha, [])),
    )

    expect(fx.revParse(knot, 'refs/tags/v1')).toBe(tagSha)
    expect(fx.git(['cat-file', '-t', 'refs/tags/v1'], knot)).toBe('tag')
  })
})