/**
 * The watch loop: one firehose connection per PDS, a run for every ref that
 * moves on a watched repository and every check the owner asks for.
 *
 * Runs are idempotent under replay. The ref map records what each ref pointed
 * at when it was last handled, and the request map records the version of each
 * request acted on, so an event seen twice produces no second run.
 */

import { GIT_REPO_COLLECTION, parseRepoRecord } from '@pdsjs/git';
import { subscribeRepos } from './firehose.js';
import { GIT_CHECK_REQUEST_COLLECTION } from './lexicon.js';
import { diffRefs, refsToState } from './refs.js';
import {
  readRepoRecord as defaultReadRepoRecord,
  parseCheckRequest,
  requestedSha,
} from './request.js';
import { repoKey } from './spec.js';

/** The collections one firehose connection carries to this daemon. */
export const WATCHED_COLLECTIONS = [
  GIT_REPO_COLLECTION,
  GIT_CHECK_REQUEST_COLLECTION,
];

/**
 * @typedef {Object} WatchedRepo
 * @property {string} did
 * @property {string} repoName
 * @property {string} service - PDS base URL
 */

/**
 * @typedef {Object} DaemonContext
 * @property {WatchedRepo[]} repos
 * @property {import('./runner.js').RunCheck} runCheck
 * @property {import('./state.js').StateStore} state
 * @property {AbortSignal} [signal]
 * @property {typeof defaultReadRepoRecord} [readRepoRecord] - how startup learns a repository's refs
 * @property {(repoDid: string, repoName: string, refs: string[]) => Promise<string[]>} [unfinishedRefs] - which refs the runner left mid-run
 * @property {(message: string) => void} [onNotice]
 */

/**
 * What one PDS's events mean for the repositories watched on it.
 * @param {Object} ctx
 * @param {string} ctx.service
 * @param {WatchedRepo[]} ctx.repos - all on this service
 * @param {import('./state.js').StateStore} ctx.state
 * @param {import('./runner.js').RunCheck} ctx.runCheck
 * @param {typeof defaultReadRepoRecord} [ctx.readRepoRecord] - how a requested run learns the repository's current version
 * @param {(message: string) => void} [ctx.onNotice]
 * @returns {(write: import('./firehose.js').RecordWrite) => Promise<void>}
 */
export function createWriteHandler(ctx) {
  const { service, state, runCheck } = ctx;
  const readRepoRecord = ctx.readRepoRecord ?? defaultReadRepoRecord;
  const onNotice = ctx.onNotice ?? (() => {});
  const watched = new Set(
    ctx.repos.map((repo) => repoKey(repo.did, repo.repoName)),
  );
  const watchedDids = new Set(ctx.repos.map((repo) => repo.did));

  /**
   * A push: run every ref that moved.
   * @param {import('./firehose.js').RecordWrite} write
   */
  async function onRepoWrite(write) {
    // The record carries the repository's name; its key says nothing about
    // what it is called, so the watch list is matched on the name.
    const record = parseRepoRecord(write.value);
    const key = repoKey(write.did, record.name);
    if (!watched.has(key)) return;

    for (const change of diffRefs(state.refs(key), record.refs)) {
      await runCheck({
        service,
        did: write.did,
        repoName: record.name,
        repoRkey: write.rkey,
        subject: { uri: write.uri, cid: write.cid },
        ref: change.name,
        sha: change.sha,
      });
    }
    // The ref map advances after the runs. A crash mid-run leaves it and the
    // cursor behind the event, so the event replays and the run happens once
    // the daemon is back.
    state.setRefs(key, refsToState(record.refs));
  }

  /**
   * A run the owner asked for. The record's authority is the authorisation:
   * only the account that owns a watched repository can name it here, because
   * the request is a record in that account's own repo.
   *
   * The ref map is left alone. A requested run says nothing about where the
   * ref points, so a later push of the same commit still starts no run.
   * @param {import('./firehose.js').RecordWrite} write
   */
  async function onRequestWrite(write) {
    const request = parseCheckRequest(write.value);
    if (!request) return;
    // An account this runner watches nothing of is answered before any read.
    if (!watchedDids.has(write.did)) return;

    // The request names the repository by its record key, so the record is
    // read before the watch list can be matched against its name.
    /** @type {import('./request.js').RepoRecordVersion} */
    let record;
    try {
      record = await readRepoRecord({
        service,
        did: write.did,
        rkey: request.repo,
      });
    } catch {
      // A request naming a repository this server does not hold is nothing
      // this runner can act on.
      return;
    }
    const key = repoKey(write.did, record.name);
    if (!watched.has(key)) return;
    if (state.request(key, write.rkey) === write.cid) return;

    const sha = requestedSha(request, record.refs);
    if (!sha) {
      onNotice(`${record.name}: ${request.ref} is not a ref of this repo`);
      // Marked handled: replaying the request will not make the ref appear.
      state.setRequest(key, write.rkey, write.cid);
      return;
    }

    onNotice(`${record.name}: ${request.ref} requested`);
    await runCheck({
      service,
      did: write.did,
      repoName: record.name,
      repoRkey: request.repo,
      subject: { uri: record.uri, cid: record.cid },
      ref: request.ref,
      sha,
    });
    state.setRequest(key, write.rkey, write.cid);
  }

  return async function onWrite(write) {
    if (write.collection === GIT_CHECK_REQUEST_COLLECTION) {
      await onRequestWrite(write);
      return;
    }
    await onRepoWrite(write);
  };
}

/**
 * Watch every repository until the signal aborts.
 * @param {DaemonContext} ctx
 * @returns {Promise<void>}
 */
export async function watch(ctx) {
  const { repos, runCheck, state, signal } = ctx;
  const onNotice = ctx.onNotice ?? (() => {});
  const readRepoRecord = ctx.readRepoRecord ?? defaultReadRepoRecord;
  const unfinishedRefs = ctx.unfinishedRefs;

  // A repository the state file says nothing about has no ref this daemon has
  // seen, so the first event would read every ref in the record as new and
  // start a run for each, tags included. Reading the record once here makes
  // the first run come from a ref that moves after startup, which is what
  // starting at the live head means. A repository that cannot be read is left
  // unseeded: the daemon still watches it, and the first event runs it.
  //
  // A ref whose latest check is still `running` is left unadopted. Its run
  // was cut off with the check record half written, and adopting the ref
  // would leave that record `running` for good; unadopted, the ref runs again
  // on the next event for the repository.
  for (const repo of repos) {
    const key = repoKey(repo.did, repo.repoName);
    if (Object.keys(state.refs(key)).length > 0) continue;
    try {
      const record = await readRepoRecord(repo);
      let refs = record.refs;
      if (unfinishedRefs) {
        const cut = await unfinishedRefs(
          repo.did,
          record.rkey,
          refs.map((ref) => ref.name),
        );
        if (cut.length > 0) {
          refs = refs.filter((ref) => !cut.includes(ref.name));
          onNotice(`${repo.repoName}: ${cut.join(', ')} was left mid-run`);
        }
      }
      state.setRefs(key, refsToState(refs));
      onNotice(`${repo.repoName}: starting from ${refs.length} refs`);
    } catch (err) {
      onNotice(
        `${repo.repoName}: could not read its refs (${err instanceof Error ? err.message : err})`,
      );
    }
  }

  /** @type {Map<string, WatchedRepo[]>} */
  const byService = new Map();
  for (const repo of repos) {
    const group = byService.get(repo.service) ?? [];
    group.push(repo);
    byService.set(repo.service, group);
  }

  const connections = [...byService.entries()].map(([service, group]) =>
    subscribeRepos({
      service,
      collections: WATCHED_COLLECTIONS,
      cursor: state.cursor(service),
      signal,
      onNotice,
      onSeq: (seq) => state.setCursor(service, seq),
      onWrite: createWriteHandler({
        service,
        repos: group,
        state,
        runCheck,
        onNotice,
      }),
    }),
  );

  onNotice(
    `watching ${repos.length} repositor${repos.length === 1 ? 'y' : 'ies'} on ${byService.size} host${byService.size === 1 ? '' : 's'}`,
  );
  await Promise.all(connections);
}