// @pdsjs/spaces/outbox - notices this server owes other services, persisted
// before they are sent and retried until they land.
//
// Two kinds. As a repo host, this server owes the authority of each space it
// writes into a notifyWrite for the repo's latest rev: the authority's writer
// set, and with it listRepos, is built from these notices. As an authority, it
// owes every syncer registered for a space it deleted a notifySpaceDeleted.
//
// A notice that fails for a reason that may pass is retried with backoff. A
// write notice stops when it lands, when a newer write to the same space
// replaces it, or a day after its write. A deletion notice stops when it
// lands, when the syncer's registration expires, or when the space exists
// again.

import { spaceHostAudience, spaceHostEndpoint } from './authority.js';
import { createServiceAuth } from './service-auth.js';
import { parseSpaceUri } from './uri.js';

/** How long one repo rev is retried before delivery stops. */
export const DELIVERY_DEADLINE_MS = 24 * 60 * 60 * 1000;
const FIRST_RETRY_MS = 10 * 1000;
const MAX_RETRY_MS = 60 * 60 * 1000;
const BATCH = 50;
/** How long one delivery attempt waits for an answer. */
export const NOTICE_TIMEOUT_MS = 3000;

/**
 * The wait before retry number `attempts`, counted from 1.
 * @param {number} attempts
 * @returns {number}
 */
export function retryDelay(attempts) {
  return Math.min(FIRST_RETRY_MS * 2 ** (attempts - 1), MAX_RETRY_MS);
}

/**
 * Whether a response says the same request may succeed later.
 * @param {number} status
 * @returns {boolean}
 */
function isRetryable(status) {
  return status === 408 || status === 425 || status === 429 || status >= 500;
}

/**
 * @param {Object} ctx
 * @param {import('@pdsjs/core/ports').SpaceStoragePort} ctx.spaceStorage
 * @param {() => Promise<string|null>} ctx.getDid
 * @param {() => Promise<import('./token.js').SpaceSigner>} ctx.getSigner
 * @param {import('@pdsjs/core/ports').DidResolverPort} ctx.resolveDid
 * @param {typeof fetch} [ctx.fetch]
 * @param {(at: number) => Promise<void>} [ctx.scheduleWakeup]
 * @param {(work: Promise<unknown>) => void} [ctx.background] - runs the first
 *   delivery attempt after the response; absent, the attempt runs before it
 * @param {() => number} [ctx.now] - epoch ms
 */
export function createNotifyOutbox(ctx) {
  const now = ctx.now ?? Date.now;

  /**
   * POST a notice, signed as `iss` for `aud`.
   * @param {string} url
   * @param {{iss: string, aud: string, lxm: string}} auth
   * @param {Record<string, unknown>} body
   * @returns {Promise<'delivered'|'retry'|'failed'>}
   */
  async function post(url, { iss, aud, lxm }, body) {
    try {
      const token = await createServiceAuth({
        iss,
        aud,
        lxm,
        signer: await ctx.getSigner(),
      });
      const res = await (ctx.fetch ?? fetch)(url, {
        method: 'POST',
        headers: {
          'content-type': 'application/json',
          authorization: `Bearer ${token}`,
        },
        body: JSON.stringify(body),
        signal: AbortSignal.timeout(NOTICE_TIMEOUT_MS),
      });
      if (res.ok) return 'delivered';
      return isRetryable(res.status) ? 'retry' : 'failed';
    } catch {
      return 'retry';
    }
  }

  /**
   * One delivery attempt of a write notice.
   * @param {string} space
   * @param {string} repoRev
   * @param {Uint8Array} data
   * @returns {Promise<'delivered'|'retry'|'failed'>}
   */
  async function sendWrite(space, repoRev, data) {
    const did = await ctx.getDid();
    if (!did) return 'failed';
    /** @type {string} */
    let spaceDid;
    try {
      ({ spaceDid } = parseSpaceUri(space));
    } catch {
      return 'failed';
    }
    /** @type {string} */
    let endpoint;
    try {
      const didDoc = await ctx.resolveDid(spaceDid);
      if (!didDoc) return 'retry';
      endpoint = spaceHostEndpoint(didDoc, spaceDid);
    } catch {
      // A DID document naming no PDS yet.
      return 'retry';
    }
    return post(
      `${endpoint}/xrpc/com.atproto.space.notifyWrite`,
      {
        iss: did,
        aud: spaceHostAudience(spaceDid),
        lxm: 'com.atproto.space.notifyWrite',
      },
      { space, repo: did, repoRev, data: { $bytes: toBase64(data) } },
    );
  }

  /**
   * One delivery attempt of a deletion notice.
   * @param {string} space
   * @param {string} service - the syncer's service identifier
   * @param {string} serviceEndpoint
   * @returns {Promise<'delivered'|'retry'|'failed'>}
   */
  async function sendDeleted(space, service, serviceEndpoint) {
    const did = await ctx.getDid();
    if (!did) return 'failed';
    return post(
      `${serviceEndpoint}/xrpc/com.atproto.space.notifySpaceDeleted`,
      { iss: did, aud: service, lxm: 'com.atproto.space.notifySpaceDeleted' },
      { space },
    );
  }

  /**
   * When the next retry of a notice that has failed `attempts` times falls,
   * or null once that is past its deadline.
   * @param {number} attempts
   * @param {string} deadlineAt
   * @returns {number|null}
   */
  function nextAttempt(attempts, deadlineAt) {
    const at = now() + retryDelay(attempts);
    return at <= Date.parse(deadlineAt) ? at : null;
  }

  /**
   * Act on a write notice's outcome. `attempts` counts the failures before it.
   * @param {import('@pdsjs/core/ports').SpaceNotifyRow} row
   * @param {'delivered'|'retry'|'failed'} outcome
   * @returns {Promise<void>}
   */
  async function settleWrite(row, outcome) {
    if (outcome === 'retry') {
      const attempts = row.attempts + 1;
      const at = nextAttempt(attempts, row.deadlineAt);
      if (at !== null) {
        await ctx.spaceStorage.rescheduleSpaceNotify(
          row.space,
          row.repoRev,
          attempts,
          new Date(at).toISOString(),
        );
        await ctx.scheduleWakeup?.(at);
        return;
      }
    }
    await ctx.spaceStorage.deleteSpaceNotify(row.space, row.repoRev);
  }

  /**
   * Act on a deletion notice's outcome.
   * @param {import('@pdsjs/core/ports').SpaceDeletedNotifyRow} row
   * @param {'delivered'|'retry'|'failed'} outcome
   * @returns {Promise<void>}
   */
  async function settleDeleted(row, outcome) {
    if (outcome === 'retry') {
      const attempts = row.attempts + 1;
      // The last retry falls on the deadline itself.
      const at = Math.min(
        now() + retryDelay(attempts),
        Date.parse(row.deadlineAt),
      );
      if (at > now()) {
        await ctx.spaceStorage.rescheduleSpaceDeletedNotify(
          row.space,
          row.service,
          attempts,
          new Date(at).toISOString(),
        );
        await ctx.scheduleWakeup?.(at);
        return;
      }
    }
    await ctx.spaceStorage.deleteSpaceDeletedNotify(row.space, row.service);
  }

  /**
   * Retry a deletion notice that is due. A space that exists again is not
   * deleted any more, and the notice would purge the syncer's new copy.
   * @param {import('@pdsjs/core/ports').SpaceDeletedNotifyRow} row
   * @returns {Promise<void>}
   */
  async function retryDeleted(row) {
    const current = await ctx.spaceStorage.getSpace(row.space);
    if (!current?.deletedAt || Date.parse(row.deadlineAt) < now()) {
      await ctx.spaceStorage.deleteSpaceDeletedNotify(row.space, row.service);
      return;
    }
    await settleDeleted(
      row,
      await sendDeleted(row.space, row.service, row.serviceEndpoint),
    );
  }

  return {
    /**
     * Owe a space's authority this write, and try to deliver it now. A newer
     * write to the same space replaces an older undelivered one and starts its
     * retries and deadline afresh.
     * @param {string} space
     * @param {{rev: string, data: Uint8Array}} commit
     * @returns {Promise<void>}
     */
    async deliver(space, { rev, data }) {
      const at = now();
      /** @type {import('@pdsjs/core/ports').SpaceNotifyRow} */
      const row = {
        space,
        repoRev: rev,
        data,
        attempts: 0,
        nextAttemptAt: new Date(at).toISOString(),
        deadlineAt: new Date(at + DELIVERY_DEADLINE_MS).toISOString(),
      };
      await ctx.spaceStorage.enqueueSpaceNotify(row);
      const attempt = sendWrite(space, rev, data).then((outcome) =>
        settleWrite(row, outcome),
      );
      if (ctx.background) ctx.background(attempt);
      else await attempt;
    },

    /**
     * Owe every registered syncer a notice that this space is deleted, and
     * try each now. A syncer is owed it until its registration expires: after
     * that it must renew its credential, and is answered SpaceDeleted.
     * @param {string} space
     * @param {import('@pdsjs/core/ports').SpaceRecipientRow[]} recipients
     * @returns {Promise<void>}
     */
    async deliverSpaceDeleted(space, recipients) {
      const at = new Date(now()).toISOString();
      await Promise.all(
        recipients.map(async (recipient) => {
          /** @type {import('@pdsjs/core/ports').SpaceDeletedNotifyRow} */
          const row = {
            space,
            service: recipient.serviceDid,
            serviceEndpoint: recipient.serviceEndpoint,
            attempts: 0,
            nextAttemptAt: at,
            deadlineAt: recipient.expiresAt,
          };
          await ctx.spaceStorage.enqueueSpaceDeletedNotify(row);
          await settleDeleted(
            row,
            await sendDeleted(space, row.service, row.serviceEndpoint),
          );
        }),
      );
    },

    /**
     * Retry every notice that is due.
     * @returns {Promise<number|null>} when the next one falls due, or null
     */
    async tick() {
      const nowIso = new Date(now()).toISOString();
      const due = await ctx.spaceStorage.listDueSpaceNotify(nowIso, BATCH);
      for (const row of due) {
        await settleWrite(
          row,
          await sendWrite(row.space, row.repoRev, row.data),
        );
      }
      const deleted = await ctx.spaceStorage.listDueSpaceDeletedNotify(
        nowIso,
        BATCH,
      );
      for (const row of deleted) await retryDeleted(row);
      const next = await ctx.spaceStorage.nextSpaceNotifyAt();
      return next === null ? null : Date.parse(next);
    },
  };
}

/**
 * @param {Uint8Array} bytes
 * @returns {string}
 */
function toBase64(bytes) {
  let binary = '';
  for (const b of bytes) binary += String.fromCharCode(b);
  return btoa(binary).replace(/=+$/, '');
}