/**
 * Writing check records into the runner's own repo.
 *
 * A run publishes twice: `running` when the steps start, then the outcome
 * under the same record key with a compare-and-swap. A reader that sees only
 * the first write knows the runner stopped mid-run.
 */

import { createTid } from '@pdsjs/core/repo';
import { XrpcError } from '@pdsjs/git';
import {
  GIT_CHECK_COLLECTION,
  GIT_LATEST_CHECK_COLLECTION,
} from './lexicon.js';
import { latestCheckKey } from './spec.js';

/** Log bytes kept on a check record. */
const DEFAULT_MAX_LOG_BYTES = 1024 * 1024;

/** Attempts at one PDS write before the failure reaches the runner. */
const WRITE_ATTEMPTS = 3;

/** Pause before a second attempt; the next pause doubles it. */
const DEFAULT_RETRY_DELAY_MS = 1000;

const TRUNCATION_NOTICE = '[earlier output dropped, log too large]\n';

/**
 * Keep the tail of a log, which is where a failing step reports. The notice
 * replaces what was dropped, so the blob never exceeds the cap.
 * @param {string} text
 * @param {number} maxBytes
 * @returns {Uint8Array}
 */
export function truncateLogs(text, maxBytes) {
  const bytes = new TextEncoder().encode(text);
  if (bytes.length <= maxBytes) return bytes;
  const notice = new TextEncoder().encode(TRUNCATION_NOTICE);
  const tail = bytes.subarray(bytes.length - (maxBytes - notice.length));
  const out = new Uint8Array(notice.length + tail.length);
  out.set(notice, 0);
  out.set(tail, notice.length);
  return out;
}

/**
 * The outcome a set of step results describes. Fewer results than steps means
 * the executor stopped early, which it only does after a failure.
 * @param {import('./executor.js').StepResult[]} results
 * @returns {'success'|'failure'}
 */
export function checkStatus(results) {
  return results.every((result) => result.exitCode === 0)
    ? 'success'
    : 'failure';
}

/**
 * @typedef {Object} CheckSubject
 * @property {string} uri
 * @property {string} cid
 */

/**
 * @typedef {Object} LatestCheckEntry
 * @property {string} workflow
 * @property {CheckSubject} subject
 * @property {CheckSubject} check
 * @property {string} sha
 * @property {string} status
 * @property {string} startedAt
 * @property {string} [finishedAt]
 */

/**
 * Which of a repository's refs the runner last left mid-run.
 *
 * The check records are the runner's own memory of what it was doing, and
 * they outlive its state file. A daemon that starts with no state adopts the
 * refs it finds so it does not rerun the whole record; a ref whose latest
 * check is still `running` is the one exception, because the run it belongs
 * to was cut off and nothing else will finish it.
 * @param {Object} ctx
 * @param {import('@pdsjs/git').XrpcClient} ctx.client - the runner's PDS
 * @param {string} ctx.did - the runner's own DID
 * @returns {(repoDid: string, repoRkey: string, refs: string[]) => Promise<string[]>}
 */
export function createUnfinishedRefsReader(ctx) {
  const { client, did } = ctx;
  return async (repoDid, repoRkey, refs) => {
    /** @type {string[]} */
    const unfinished = [];
    for (const ref of refs) {
      const rkey = latestCheckKey(repoDid, repoRkey, ref);
      // A repository the runner has never published for reads as absent,
      // which is not a cut-off run.
      const current = await client.getRecord(
        did,
        GIT_LATEST_CHECK_COLLECTION,
        rkey,
      );
      const entries = latestCheckEntries(current?.value);
      if (entries.some((entry) => entry.status === 'running')) {
        unfinished.push(ref);
      }
    }
    return unfinished;
  };
}

/**
 * The per-workflow entries of a latest-check record. A record written
 * before entries existed carries one workflow's fields at the top level;
 * it reads as one entry.
 * @param {unknown} value
 * @returns {LatestCheckEntry[]}
 */
export function latestCheckEntries(value) {
  if (!value || typeof value !== 'object') return [];
  const record = /** @type {Record<string, unknown>} */ (value);
  if (Array.isArray(record.checks)) {
    return /** @type {LatestCheckEntry[]} */ (record.checks);
  }
  if (typeof record.status !== 'string') return [];
  const single = /** @type {LatestCheckEntry} */ ({
    workflow: typeof record.workflow === 'string' ? record.workflow : 'ci',
    subject: record.subject,
    check: record.check,
    sha: record.sha,
    status: record.status,
    startedAt: record.startedAt,
    ...(record.finishedAt ? { finishedAt: record.finishedAt } : {}),
  });
  return [single];
}

/**
 * @typedef {Object} PublisherContext
 * @property {import('@pdsjs/git').XrpcClient} client - authenticated as the runner
 * @property {string} did - the runner's DID
 * @property {number} [maxLogBytes]
 * @property {number} [retryDelayMs]
 */

/** @typedef {ReturnType<typeof createPublisher>} Publisher */

/**
 * @param {PublisherContext} ctx
 */
export function createPublisher(ctx) {
  const { client, did } = ctx;
  const maxLogBytes = ctx.maxLogBytes ?? DEFAULT_MAX_LOG_BYTES;
  const retryDelayMs = ctx.retryDelayMs ?? DEFAULT_RETRY_DELAY_MS;

  /**
   * Run one PDS write, retrying a server error or a dropped connection.
   * A client error names a permanent condition (a failed swap, a rejected
   * record) and throws at once. Without the retries, one transient failure
   * on the outcome write leaves the check `running` forever, and a reader
   * cannot tell it from a runner that died mid-run.
   * @template T
   * @param {() => Promise<T>} write
   * @returns {Promise<T>}
   */
  async function withRetries(write) {
    for (let attempt = 1; ; attempt++) {
      try {
        return await write();
      } catch (err) {
        const permanent =
          err instanceof XrpcError && err.status > 0 && err.status < 500;
        if (permanent || attempt === WRITE_ATTEMPTS) throw err;
        await new Promise((resolve) =>
          setTimeout(resolve, retryDelayMs * attempt),
        );
      }
    }
  }

  /**
   * Update this workflow's entry in the repository's latest-check record.
   *
   * The record holds one entry per workflow, so a run reads the record,
   * replaces its own entry and writes the rest back unchanged. The swap
   * condition guards the read-modify-write; a lost swap re-reads whoever
   * won. One daemon runs serially, so the loop settles at once in practice.
   * @param {Object} run
   * @param {string} run.repoDid
   * @param {string} run.repoName - what the repository is called
   * @param {string} run.repoRkey - the repository's record key
   * @param {CheckSubject} run.subject
   * @param {string} run.ref
   * @param {string} run.sha
   * @param {string} run.workflow
   * @param {string} run.startedAt
   * @param {CheckSubject} check
   * @param {'running'|'success'|'failure'|'error'} status
   * @param {string} [finishedAt]
   * @returns {Promise<void>}
   */
  async function writeLatest(run, check, status, finishedAt) {
    const rkey = latestCheckKey(run.repoDid, run.repoRkey, run.ref);
    const entry = {
      workflow: run.workflow,
      subject: run.subject,
      check,
      sha: run.sha,
      status,
      startedAt: run.startedAt,
      ...(finishedAt ? { finishedAt } : {}),
    };
    for (let attempt = 1; ; attempt++) {
      const current = await withRetries(() =>
        client.getRecord(did, GIT_LATEST_CHECK_COLLECTION, rkey),
      );
      const others = latestCheckEntries(current?.value).filter(
        (existing) => existing.workflow !== run.workflow,
      );
      try {
        await withRetries(() =>
          client.putRecord(
            did,
            GIT_LATEST_CHECK_COLLECTION,
            rkey,
            {
              $type: GIT_LATEST_CHECK_COLLECTION,
              ref: run.ref,
              checks: [...others, entry],
            },
            current ? current.cid : null,
          ),
        );
        return;
      } catch (err) {
        const lostSwap =
          err instanceof XrpcError && err.error === 'InvalidSwap';
        if (!lostSwap || attempt === WRITE_ATTEMPTS) throw err;
      }
    }
  }

  return {
    /**
     * @param {Object} run
     * @param {string} run.repoDid
     * @param {string} run.repoName - what the repository is called
     * @param {string} run.repoRkey - the repository's record key
     * @param {CheckSubject} run.subject
     * @param {string} run.ref
     * @param {string} run.sha
     * @param {string} run.workflow
     * @param {string} run.startedAt
     * @returns {Promise<{rkey: string, cid: string}>}
     */
    async start(run) {
      const rkey = createTid();
      const { uri, cid } = await withRetries(() =>
        client.putRecord(
          did,
          GIT_CHECK_COLLECTION,
          rkey,
          {
            $type: GIT_CHECK_COLLECTION,
            subject: run.subject,
            ref: run.ref,
            sha: run.sha,
            workflow: run.workflow,
            status: 'running',
            startedAt: run.startedAt,
          },
          null,
        ),
      );
      await writeLatest(run, { uri, cid }, 'running');
      return { rkey, cid };
    },

    /**
     * @param {Object} run
     * @param {string} run.rkey
     * @param {string} run.cid - the record version this replaces
     * @param {string} run.repoDid
     * @param {string} run.repoName - what the repository is called
     * @param {string} run.repoRkey - the repository's record key
     * @param {CheckSubject} run.subject
     * @param {string} run.ref
     * @param {string} run.sha
     * @param {string} run.workflow
     * @param {'success'|'failure'|'error'} run.status
     * @param {import('./executor.js').StepResult[]} run.steps
     * @param {string} run.startedAt
     * @param {string} run.logs
     * @returns {Promise<string>} the record uri
     */
    async finish(run) {
      const bytes = truncateLogs(run.logs, maxLogBytes);
      const logs = await withRetries(() =>
        client.uploadBlob(bytes, 'text/plain'),
      );
      const finishedAt = new Date().toISOString();
      const { uri, cid } = await withRetries(() =>
        client.putRecord(
          did,
          GIT_CHECK_COLLECTION,
          run.rkey,
          {
            $type: GIT_CHECK_COLLECTION,
            subject: run.subject,
            ref: run.ref,
            sha: run.sha,
            workflow: run.workflow,
            status: run.status,
            steps: run.steps,
            logs,
            startedAt: run.startedAt,
            finishedAt,
          },
          run.cid,
        ),
      );
      await writeLatest(run, { uri, cid }, run.status, finishedAt);
      return uri;
    },
  };
}