import { writeFileSync } from 'node:fs';
import { mkdtemp, rm } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { createWriteHandler, watch } from '../src/daemon.js';
import { createStateStore, statePath } from '../src/state.js';

const SERVICE = 'https://pds.example.com';
const DID = 'did:plc:abc';
const REPOS = [{ did: DID, repoName: 'proj', service: SERVICE }];

/** @type {string} */
let dir;

beforeEach(async () => {
  dir = await mkdtemp(join(tmpdir(), 'git-ci-state-'));
});

afterEach(async () => {
  await rm(dir, { recursive: true, force: true });
});

const MAIN = 'a'.repeat(40);

/** The repository's record key, which is not what it is called. */
const RKEY = '3lprojkey';

/**
 * @param {Array<{name: string, sha: string}>} refs
 * @param {string} [name]
 * @param {string} [rkey]
 */
function write(refs, name = 'proj', rkey = RKEY) {
  return {
    did: DID,
    collection: 'dev.pdsjs.git.repo',
    rkey,
    cid: 'bafyreiaaa',
    uri: `at://${DID}/dev.pdsjs.git.repo/${rkey}`,
    seq: 1,
    value: {
      name,
      refs,
      bundles: [],
      createdAt: '2026-08-18T00:00:00.000Z',
    },
  };
}

/**
 * @param {Record<string, unknown>} value
 * @param {Object} [opts]
 * @param {string} [opts.rkey]
 * @param {string} [opts.cid]
 * @param {string} [opts.did]
 */
function requestWrite(value, opts = {}) {
  const rkey = opts.rkey ?? 'proj:refs_heads_main';
  const did = opts.did ?? DID;
  return {
    did,
    collection: 'dev.pdsjs.git.checkRequest',
    rkey,
    cid: opts.cid ?? 'bafyreireq1',
    uri: `at://${did}/dev.pdsjs.git.checkRequest/${rkey}`,
    seq: 2,
    value,
  };
}

/** @param {{refs?: Array<{name: string, sha: string}>}} [opts] */
function harness(opts = {}) {
  const state = createStateStore(join(dir, 'state.json'));
  /** @type {Array<Record<string, *>>} */
  const runs = [];
  /** @type {string[]} */
  const notices = [];
  /** @type {Array<Record<string, *>>} */
  const reads = [];
  const onWrite = createWriteHandler({
    service: SERVICE,
    repos: REPOS,
    state,
    runCheck: async (/** @type {Record<string, *>} */ request) => {
      runs.push(request);
      return 'success';
    },
    readRepoRecord: async (/** @type {Record<string, *>} */ target) => {
      reads.push(target);
      if (target.rkey && target.rkey !== RKEY) {
        throw new Error(`no repository at ${target.rkey}`);
      }
      return {
        uri: `at://${DID}/dev.pdsjs.git.repo/${RKEY}`,
        cid: 'bafyreicurrent',
        name: 'proj',
        rkey: RKEY,
        refs: opts.refs ?? [{ name: 'refs/heads/main', sha: MAIN }],
      };
    },
    onNotice: (/** @type {string} */ message) => notices.push(message),
  });
  return { onWrite, runs, state, notices, reads };
}

describe('createWriteHandler', () => {
  it('runs the ref that moved', async () => {
    const { onWrite, runs } = harness();
    await onWrite(write([{ name: 'refs/heads/main', sha: 'a'.repeat(40) }]));

    expect(runs).toHaveLength(1);
    expect(runs[0].ref).toBe('refs/heads/main');
    expect(runs[0].sha).toBe('a'.repeat(40));
    expect(runs[0].repoName).toBe('proj');
    expect(runs[0].service).toBe(SERVICE);
    expect(runs[0].subject).toEqual({
      uri: `at://${DID}/dev.pdsjs.git.repo/${RKEY}`,
      cid: 'bafyreiaaa',
    });
  });

  it('ignores a repository it does not watch', async () => {
    const { onWrite, runs } = harness();
    await onWrite(
      write([{ name: 'refs/heads/main', sha: 'a'.repeat(40) }], 'other'),
    );
    expect(runs).toEqual([]);
  });

  it('does not re-run a ref that did not move', async () => {
    const { onWrite, runs } = harness();
    const event = write([{ name: 'refs/heads/main', sha: 'a'.repeat(40) }]);
    await onWrite(event);
    await onWrite(event);
    expect(runs).toHaveLength(1);
  });

  it('runs once per ref that moved in the same push', async () => {
    const { onWrite, runs } = harness();
    await onWrite(
      write([
        { name: 'refs/heads/main', sha: 'a'.repeat(40) },
        { name: 'refs/heads/topic', sha: 'b'.repeat(40) },
      ]),
    );
    expect(runs.map((run) => run.ref)).toEqual([
      'refs/heads/main',
      'refs/heads/topic',
    ]);
  });

  it('leaves the ref map behind when a run throws, so the event replays', async () => {
    const state = createStateStore(join(dir, 'state.json'));
    let attempts = 0;
    const onWrite = createWriteHandler({
      service: SERVICE,
      repos: REPOS,
      state,
      runCheck: async () => {
        attempts += 1;
        if (attempts === 1) throw new Error('runner died');
        return 'success';
      },
    });
    const event = write([{ name: 'refs/heads/main', sha: 'a'.repeat(40) }]);

    await expect(onWrite(event)).rejects.toThrow('runner died');
    await onWrite(event);
    expect(attempts).toBe(2);
  });
});

describe('createWriteHandler, on a check request', () => {
  const REQUEST = {
    repo: RKEY,
    ref: 'refs/heads/main',
    requestedAt: '2026-08-19T00:00:00.000Z',
  };

  it('runs the ref at the commit the repository record names now', async () => {
    const { onWrite, runs, reads } = harness();
    await onWrite(requestWrite(REQUEST));

    expect(reads).toEqual([{ service: SERVICE, did: DID, rkey: RKEY }]);
    expect(runs).toHaveLength(1);
    expect(runs[0].ref).toBe('refs/heads/main');
    expect(runs[0].sha).toBe(MAIN);
    expect(runs[0].subject).toEqual({
      uri: `at://${DID}/dev.pdsjs.git.repo/${RKEY}`,
      cid: 'bafyreicurrent',
    });
  });

  it('runs the commit the request names, not the ref tip', async () => {
    const { onWrite, runs } = harness();
    const sha = 'b'.repeat(40);
    await onWrite(requestWrite({ ...REQUEST, sha }));
    expect(runs[0].sha).toBe(sha);
  });

  it('ignores a request naming a repository it does not watch', async () => {
    const { onWrite, runs, reads } = harness();
    await onWrite(requestWrite({ ...REQUEST, repo: '3lotherkey' }));
    await onWrite(requestWrite(REQUEST, { did: 'did:plc:someoneelse' }));
    expect(runs).toEqual([]);
    // An account nothing is watched of costs no read at all; a key this
    // account does not hold costs the one read that says so.
    expect(reads).toEqual([{ service: SERVICE, did: DID, rkey: '3lotherkey' }]);
  });

  it('ignores a request for a repository of a name it does not watch', async () => {
    /** @type {Array<Record<string, *>>} */
    const runs = [];
    const onWrite = createWriteHandler({
      service: SERVICE,
      repos: REPOS,
      state: createStateStore(join(dir, 'state.json')),
      runCheck: async (/** @type {Record<string, *>} */ request) => {
        runs.push(request);
        return 'success';
      },
      // The account holds this repository; the runner watches another of
      // its repositories, and the name is what says so.
      readRepoRecord: async () => ({
        uri: `at://${DID}/dev.pdsjs.git.repo/3lotherkey`,
        cid: 'bafyreicurrent',
        name: 'unwatched',
        rkey: '3lotherkey',
        refs: [{ name: 'refs/heads/main', sha: MAIN }],
      }),
    });

    await onWrite(requestWrite({ ...REQUEST, repo: '3lotherkey' }));
    expect(runs).toEqual([]);
  });

  it('ignores a request missing the repository or the ref', async () => {
    const { onWrite, runs } = harness();
    await onWrite(requestWrite({ ref: 'refs/heads/main' }));
    await onWrite(requestWrite({ repo: RKEY }));
    await onWrite(requestWrite({ repo: RKEY, ref: '' }));
    expect(runs).toEqual([]);
  });

  it('runs the same request version only once', async () => {
    const { onWrite, runs } = harness();
    await onWrite(requestWrite(REQUEST));
    await onWrite(requestWrite(REQUEST));
    expect(runs).toHaveLength(1);
  });

  it('runs again when the request is made afresh', async () => {
    const { onWrite, runs } = harness();
    await onWrite(requestWrite(REQUEST));
    await onWrite(
      requestWrite(
        { ...REQUEST, requestedAt: '2026-08-19T01:00:00.000Z' },
        { cid: 'bafyreireq2' },
      ),
    );
    expect(runs).toHaveLength(2);
  });

  it('leaves the ref map alone, so the next push still starts a run', async () => {
    const { onWrite, runs, state } = harness();
    await onWrite(requestWrite(REQUEST));
    expect(state.refs('did:plc:abc/proj')).toEqual({});

    await onWrite(write([{ name: 'refs/heads/main', sha: MAIN }]));
    expect(runs).toHaveLength(2);
  });

  it('leaves the request unrecorded when the run throws, so it replays', async () => {
    const state = createStateStore(join(dir, 'state.json'));
    let attempts = 0;
    const onWrite = createWriteHandler({
      service: SERVICE,
      repos: REPOS,
      state,
      runCheck: async () => {
        attempts += 1;
        if (attempts === 1) throw new Error('runner died');
        return 'success';
      },
      readRepoRecord: async () => ({
        uri: `at://${DID}/dev.pdsjs.git.repo/${RKEY}`,
        cid: 'bafyreicurrent',
        name: 'proj',
        rkey: RKEY,
        refs: [{ name: 'refs/heads/main', sha: MAIN }],
      }),
    });

    await expect(onWrite(requestWrite(REQUEST))).rejects.toThrow('runner died');
    await onWrite(requestWrite(REQUEST));
    expect(attempts).toBe(2);
  });

  it('says so and gives up on a ref the repository does not have', async () => {
    const { onWrite, runs, notices } = harness();
    await onWrite(requestWrite({ ...REQUEST, ref: 'refs/heads/gone' }));
    expect(runs).toEqual([]);
    expect(notices.at(-1)).toMatch(/refs\/heads\/gone is not a ref/);

    // Marked handled, so the same request replaying costs no second read.
    await onWrite(requestWrite({ ...REQUEST, ref: 'refs/heads/gone' }));
    expect(notices).toHaveLength(1);
  });

  it('reads the repository over XRPC when the context names no reader', async () => {
    const state = createStateStore(join(dir, 'state.json'));
    /** @type {string[]} */
    const asked = [];
    const original = globalThis.fetch;
    globalThis.fetch = async (/** @type {any} */ url) => {
      asked.push(String(url));
      return Response.json({
        uri: `at://${DID}/dev.pdsjs.git.repo/${RKEY}`,
        cid: 'bafyreicurrent',
        value: {
          name: 'proj',
          refs: [],
          bundles: [],
          createdAt: '2026-08-18T00:00:00.000Z',
        },
      });
    };
    try {
      // No onNotice either: the ref is absent, which is the branch that
      // reports, and the default callback has to receive it silently.
      const onWrite = createWriteHandler({
        service: SERVICE,
        repos: REPOS,
        state,
        runCheck: async () => 'success',
      });
      await onWrite(requestWrite(REQUEST));
    } finally {
      globalThis.fetch = original;
    }
    expect(asked[0]).toContain('/xrpc/com.atproto.repo.getRecord');
    expect(asked[0]).toContain('collection=dev.pdsjs.git.repo');
  });
});

describe('createStateStore', () => {
  it('round-trips the cursor, the ref map and the requests through the file', () => {
    const path = join(dir, 'state.json');
    const first = createStateStore(path);
    first.setCursor(SERVICE, 99);
    first.setRefs('did:plc:abc/proj', { 'refs/heads/main': 'aa' });
    first.setRequest('did:plc:abc/proj', 'proj:refs_heads_main', 'bafyreireq');

    const second = createStateStore(path);
    expect(second.cursor(SERVICE)).toBe(99);
    expect(second.refs('did:plc:abc/proj')).toEqual({
      'refs/heads/main': 'aa',
    });
    expect(second.request('did:plc:abc/proj', 'proj:refs_heads_main')).toBe(
      'bafyreireq',
    );
  });

  it("keeps a repository's requests apart by ref", () => {
    const store = createStateStore(join(dir, 'state.json'));
    store.setRequest('did:plc:abc/proj', 'proj:refs_heads_main', 'one');
    store.setRequest('did:plc:abc/proj', 'proj:refs_heads_topic', 'two');
    expect(store.request('did:plc:abc/proj', 'proj:refs_heads_main')).toBe(
      'one',
    );
    expect(store.request('did:plc:abc/proj', 'proj:refs_heads_topic')).toBe(
      'two',
    );
  });

  // A regular file where a directory must go fails the write for every
  // uid, root included, so these run in the root containers CI uses. A
  // directory mode cannot stop root.
  it('keeps working in memory when the file cannot be written', () => {
    // A bind mount the daemon does not own reads this way, and losing
    // durability must not stop the events reaching the runner.
    const blocker = join(dir, 'blocker');
    writeFileSync(blocker, '');
    /** @type {string[]} */
    const warnings = [];
    const store = createStateStore(
      join(blocker, 'sub', 'state.json'),
      (message) => warnings.push(message),
    );

    expect(() => store.setCursor(SERVICE, 7)).not.toThrow();
    expect(() => store.setRefs('did:plc:abc/proj', { a: 'b' })).not.toThrow();
    expect(store.cursor(SERVICE)).toBe(7);
    expect(store.refs('did:plc:abc/proj')).toEqual({ a: 'b' });
    // Warned once, not once per write.
    expect(warnings).toHaveLength(1);
    expect(warnings[0]).toMatch(/not writable/);
  });

  it('carries on with no warning callback at all', () => {
    const blocker = join(dir, 'blocker-quiet');
    writeFileSync(blocker, '');
    const store = createStateStore(join(blocker, 'sub', 'state.json'));

    expect(() => store.setCursor(SERVICE, 7)).not.toThrow();
    expect(store.cursor(SERVICE)).toBe(7);
  });

  it('starts empty when the file is absent or unreadable', () => {
    const store = createStateStore(join(dir, 'missing', 'state.json'));
    expect(store.cursor(SERVICE)).toBeUndefined();
    expect(store.refs('did:plc:abc/proj')).toEqual({});
    expect(store.request('did:plc:abc/proj', 'anything')).toBeUndefined();
  });
});

describe('watch startup', () => {
  /** A firehose that connects, stays quiet, and ends when aborted. */
  const quietSignal = () => {
    const controller = new AbortController();
    queueMicrotask(() => controller.abort());
    return controller.signal;
  };

  it('adopts a repository refs before the first event', async () => {
    const store = createStateStore(join(dir, 'state.json'));
    /** @type {string[]} */
    const ran = [];
    await watch({
      repos: REPOS,
      runCheck: async (/** @type {{ref: string}} */ request) => {
        ran.push(request.ref);
        return 'success';
      },
      state: store,
      signal: quietSignal(),
      readRepoRecord: async () => ({
        uri: `at://${DID}/dev.pdsjs.git.repo/${RKEY}`,
        cid: 'bafyreiaaa',
        name: 'proj',
        rkey: RKEY,
        refs: [
          { name: 'refs/heads/main', sha: MAIN },
          { name: 'refs/tags/v1.0.0', sha: MAIN },
        ],
      }),
    });

    // Without this the next event reads both refs as new and reruns the tag,
    // which for a release workflow means publishing again.
    expect(store.refs(`${DID}/proj`)).toEqual({
      'refs/heads/main': MAIN,
      'refs/tags/v1.0.0': MAIN,
    });
    expect(ran).toEqual([]);
  });

  it('leaves a ref whose run was cut off, so it runs again', async () => {
    const store = createStateStore(join(dir, 'state.json'));
    /** @type {string[]} */
    const notices = [];
    await watch({
      repos: REPOS,
      runCheck: async () => 'success',
      state: store,
      signal: quietSignal(),
      onNotice: (message) => notices.push(message),
      readRepoRecord: async () => ({
        uri: `at://${DID}/dev.pdsjs.git.repo/${RKEY}`,
        cid: 'bafyreiaaa',
        name: 'proj',
        rkey: RKEY,
        refs: [
          { name: 'refs/heads/main', sha: MAIN },
          { name: 'refs/tags/v1.0.0', sha: MAIN },
        ],
      }),
      unfinishedRefs: async () => ['refs/heads/main'],
    });

    // Adopting main would leave its half-written check record running for
    // good. Unadopted, the next event for the repository runs it again.
    expect(store.refs(`${DID}/proj`)).toEqual({ 'refs/tags/v1.0.0': MAIN });
    expect(notices.some((n) => n.includes('was left mid-run'))).toBe(true);
  });

  it('keeps the refs it already knows', async () => {
    const store = createStateStore(join(dir, 'state.json'));
    store.setRefs(`${DID}/proj`, { 'refs/heads/main': MAIN });
    let reads = 0;
    await watch({
      repos: REPOS,
      runCheck: async () => 'success',
      state: store,
      signal: quietSignal(),
      readRepoRecord: async () => {
        reads += 1;
        return {
          uri: 'at://x',
          cid: 'bafyreiaaa',
          name: 'proj',
          rkey: RKEY,
          refs: [],
        };
      },
    });

    expect(reads).toBe(0);
    expect(store.refs(`${DID}/proj`)).toEqual({ 'refs/heads/main': MAIN });
  });

  it('watches a repository whose record it cannot read', async () => {
    const store = createStateStore(join(dir, 'state.json'));
    /** @type {string[]} */
    const notices = [];
    await watch({
      repos: REPOS,
      runCheck: async () => 'success',
      state: store,
      signal: quietSignal(),
      onNotice: (message) => notices.push(message),
      readRepoRecord: async () => {
        throw new Error('502');
      },
    });

    expect(store.refs(`${DID}/proj`)).toEqual({});
    expect(notices.some((n) => n.includes('could not read its refs'))).toBe(
      true,
    );
  });
});

describe('statePath', () => {
  it('takes an explicit path first', () => {
    expect(statePath({ PDSJS_CI_STATE: '/tmp/x.json' })).toBe('/tmp/x.json');
  });

  it('falls back to the XDG state directory', () => {
    expect(statePath({ XDG_STATE_HOME: '/state' })).toBe(
      '/state/pdsjs-git-ci/state.json',
    );
  });
});