/**
 * Serialize pull-queue drains without dropping notifications that arrive
 * while a drain is already in progress.
 */
export function createPullQueueDrain(args: {
  processRequests: () => Promise<void>;
  onError: (error: unknown) => void;
}): () => Promise<void> {
  let inFlight: Promise<void> | null = null;
  let rerunRequested = false;

  return function drain(): Promise<void> {
    if (inFlight) {
      rerunRequested = true;
      return inFlight;
    }

    inFlight = (async () => {
      try {
        do {
          rerunRequested = false;
          try {
            await args.processRequests();
          } catch (error) {
            args.onError(error);
          }
        } while (rerunRequested);
      } finally {
        inFlight = null;
      }
    })();

    return inFlight;
  };
}

type RemoteFile = { path: string; contentHash?: string; isDir: boolean };

/** Retain the latest subscription snapshot across worker/network failures.
 * Only one drain runs at a time; retry is triggered by recovery or a heartbeat. */
export function createRemoteFileQueue(args: {
  isReady: () => boolean;
  getHash: (path: string) => string | undefined;
  pull: (
    row: RemoteFile & { contentHash: string },
    shouldPublish: () => boolean,
  ) => Promise<void>;
  onError: (error: unknown, path?: string) => void;
}): {
  update: (rows: RemoteFile[]) => Promise<void>;
  retry: () => Promise<void>;
} {
  let latest = new Map<string, RemoteFile & { contentHash: string }>();
  const drain = createPullQueueDrain({
    processRequests: async () => {
      for (const row of latest.values()) {
        if (!args.isReady()) return;
        const shouldPublish = () =>
          latest.get(row.path)?.contentHash === row.contentHash;
        if (!shouldPublish() || args.getHash(row.path) === row.contentHash)
          continue;
        try {
          await args.pull(row, shouldPublish);
        } catch (error) {
          args.onError(error, row.path);
        }
      }
    },
    onError: (error) => args.onError(error),
  });
  return {
    update: (rows) => {
      latest = new Map(
        rows.flatMap((row) =>
          !row.isDir && row.contentHash
            ? [[row.path, { ...row, contentHash: row.contentHash }] as const]
            : [],
        ),
      );
      return drain();
    },
    retry: drain,
  };
}