// @pdsjs/spaces/handlers/manage - space management, the memberspace member
// list, and notifications.
//
// A space is anchored on its owner's own DID and governed by one of the
// baseline policies in ../policy.js. The member list belongs to the
// memberspace policy, and com.atproto.memberspace manages it.
//
// Notifications are best-effort and are not required for eventual consistency.
// A dropped one is recovered by a later write's notification or by a syncer's
// periodic sweep, because sync correctness rests on comparing set hashes rather
// than on receiving every event.

import { createTid } from '@pdsjs/core/repo';
import { ScopePermissions } from '@pdsjs/core/scope';
import { namedServiceEndpoint, spaceHostAudience } from '../authority.js';
import { createWriteNotifier, forwardToSyncers } from '../notify.js';
import { createNotifyOutbox } from '../outbox.js';
import {
  areMembersPublished,
  assertMemberspace,
  memberListWrites,
  memberRecordWrites,
  normalizePolicy,
  policyRecordWrites,
  SpacePolicyError,
} from '../policy.js';
import { isFutureRev, isTid, recordWriterUpdate } from '../sequence.js';
import { verifyServiceAuth } from '../service-auth.js';
import { makeSpaceRow } from '../space-row.js';
import {
  CLOCK_SKEW_SEC,
  MAX_CREDENTIAL_LIFETIME_SEC,
  SpaceTokenError,
} from '../token.js';
import { formatSpaceUri, parseSpaceUri } from '../uri.js';
import { applyWrites, MAX_WRITES_PER_COMMIT } from '../writer.js';
import { authorizeSpaceUser, createCredentialAuthorizer } from './auth.js';

// How long a write-notification registration lasts before a syncer has to
// renew it.
const REGISTRATION_TTL_MS = 24 * 60 * 60 * 1000;

// A revocation must outlive every credential it can match, allowing for clock
// skew at both ends.
const REVOCATION_RETENTION_MS =
  (MAX_CREDENTIAL_LIFETIME_SEC + 2 * CLOCK_SKEW_SEC) * 1000;

/**
 * @param {string} error
 * @param {string} message
 * @param {number} [status]
 * @returns {Response}
 */
function errorResponse(error, message, status = 400) {
  return Response.json({ error, message }, { status });
}

/**
 * @param {unknown} err
 * @returns {Response}
 */
function policyErrorResponse(err) {
  if (err instanceof SpacePolicyError) {
    return errorResponse(err.code, err.message);
  }
  throw err;
}

/**
 * @param {Request} request
 * @returns {Promise<any>}
 */
async function readJson(request) {
  try {
    return JSON.parse(await request.text());
  } catch {
    return null;
  }
}

/**
 * @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 {import('@pdsjs/core/ports').SignatureVerifierPort} ctx.verifier
 * @param {typeof fetch} [ctx.fetch]
 * @param {ReturnType<typeof createNotifyOutbox>} [ctx.outbox] - built here
 *   from the rest of ctx when absent
 * @returns {import('@pdsjs/core/pds').Routes}
 */
export function createManageRoutes(ctx) {
  const { spaceStorage, getDid, resolveDid, verifier } = ctx;
  const outbox = ctx.outbox ?? createNotifyOutbox(ctx);
  const fireNotifyWrite = createWriteNotifier(ctx, outbox);
  const requireSpaceCredential = createCredentialAuthorizer({
    spaceStorage,
    resolveDid,
    verifier,
  });

  /**
   * Management verbs are governed by the `manage=` scope param and are not
   * implied by any record action.
   * @param {{did: string, scope?: string}|null} auth
   * @param {string} space
   * @param {string} op
   * @returns {Response|null}
   */
  function checkManageScope(auth, space, op) {
    const { spaceDid, spaceType, skey } = parseSpaceUri(space);
    const permissions = new ScopePermissions(auth?.scope);
    const allowed = permissions.allowsSpaceManage({
      spaceType,
      authority: spaceDid,
      skey,
      op,
      userDid: auth?.did,
    });
    if (!allowed) {
      return errorResponse(
        'InvalidToken',
        `Scope does not permit ${op} on ${space}`,
        403,
      );
    }
    return null;
  }

  /**
   * Registrations are the authority's to hold, so this server must govern the
   * space.
   * @param {string} space
   * @returns {Promise<import('@pdsjs/core/ports').SpaceRow|Response>}
   */
  async function hostedSpace(space) {
    const row = await spaceStorage.getSpace(space);
    if (!row?.isOwner || row.deletedAt) {
      return errorResponse('SpaceNotFound', 'Space not found', 404);
    }
    return row;
  }

  /**
   * Load a space this server owns, or the Response explaining why not.
   * @param {string} space
   * @param {{did: string}|null} auth
   * @returns {Promise<import('@pdsjs/core/ports').SpaceRow|Response>}
   */
  async function ownedSpace(space, auth) {
    const row = await spaceStorage.getSpace(space);
    if (!row) return errorResponse('SpaceNotFound', 'Space not found', 404);
    if (!row.isOwner) {
      return errorResponse(
        'NotSpaceOwner',
        'This server is not the authority for that space',
        403,
      );
    }
    const hosted = await getDid();
    if (auth && auth.did !== hosted) {
      return errorResponse('NotSpaceOwner', 'Not the space owner', 403);
    }
    return row;
  }

  /**
   * Write the PDS's own records in the authority's repo in a space: the
   * published policy and member list. Applied in commits of the largest size
   * a commit allows, then notified once.
   * @param {string} space
   * @param {import('../writer.js').SpaceWriteInput[]} writes
   * @returns {Promise<void>}
   */
  async function publish(space, writes) {
    /** @type {{rev: string, setHash: Uint8Array}|null} */
    let commit = null;
    for (let i = 0; i < writes.length; i += MAX_WRITES_PER_COMMIT) {
      commit = await applyWrites(spaceStorage, {
        space,
        writes: writes.slice(i, i + MAX_WRITES_PER_COMMIT),
        owner: await getDid(),
      });
    }
    if (commit) await fireNotifyWrite(space, commit);
  }

  /**
   * The policy a request states, or the refusal.
   * @param {unknown} value
   * @returns {import('@pdsjs/core/ports').SpacePolicy|Response}
   */
  function requestedPolicy(value) {
    try {
      return normalizePolicy(value);
    } catch (err) {
      return policyErrorResponse(err);
    }
  }

  /**
   * Load a live space this server owns, governed by the memberspace policy.
   * @param {string} space
   * @param {{did: string}|null} auth
   * @returns {Promise<import('@pdsjs/core/ports').SpaceRow|Response>}
   */
  async function memberspace(space, auth) {
    const row = await ownedSpace(space, auth);
    if (row instanceof Response) return row;
    if (row.deletedAt) {
      return errorResponse('SpaceNotFound', 'Space not found', 404);
    }
    try {
      assertMemberspace(row.policy);
    } catch (err) {
      return policyErrorResponse(err);
    }
    return row;
  }

  return {
    '/xrpc/com.atproto.space.createSpace': {
      method: 'POST',
      auth: 'required',
      handler: async (request, _url, auth) => {
        const body = await readJson(request);
        if (!body) return errorResponse('InvalidRequest', 'Invalid JSON body');
        const { skey, spaceType: type } = body;
        if (typeof type !== 'string') {
          return errorResponse('InvalidRequest', 'spaceType is required');
        }
        if (!type.includes('.')) {
          return errorResponse('InvalidType', `Not an NSID: ${type}`);
        }

        // The space is anchored on the authenticated user, who becomes its
        // owner. This server hosts one account, so any other caller is a miss.
        const hosted = await getDid();
        const caller = /** @type {{did: string}} */ (auth).did;
        if (caller !== hosted) {
          return errorResponse(
            'NotSpaceOwner',
            'Can only create spaces anchored on the hosted account',
            403,
          );
        }
        if (typeof skey === 'string' && skey.length > 512) {
          return errorResponse('InvalidRequest', 'skey is too long');
        }

        const uri = formatSpaceUri({
          spaceDid: caller,
          spaceType: type,
          skey: skey || createTid(),
        });

        const scopeError = checkManageScope(auth, uri, 'create');
        if (scopeError) return scopeError;

        const policy = requestedPolicy(body.policy);
        if (policy instanceof Response) return policy;

        if (await spaceStorage.getSpace(uri)) {
          return errorResponse('SpaceAlreadyExists', `Exists: ${uri}`);
        }

        await spaceStorage.putSpace(
          makeSpaceRow(uri, { isOwner: true, policy }),
        );
        await publish(uri, await policyRecordWrites(spaceStorage, uri, policy));
        return Response.json({ uri });
      },
    },

    // Replaces the policy wholesale. A policy of another `$type` migrates the
    // space and discards the state the previous policy held, such as a
    // member list.
    '/xrpc/com.atproto.space.updateSpace': {
      method: 'POST',
      auth: 'required',
      handler: async (request, _url, auth) => {
        const body = await readJson(request);
        if (!body?.space) {
          return errorResponse('InvalidRequest', 'space is required');
        }
        const scopeError = checkManageScope(auth, body.space, 'update');
        if (scopeError) return scopeError;

        const row = await ownedSpace(body.space, auth);
        if (row instanceof Response) return row;
        if (row.deletedAt) {
          return errorResponse('SpaceNotFound', 'Space not found', 404);
        }
        const next = requestedPolicy(body.policy);
        if (next instanceof Response) return next;

        const writes = await policyRecordWrites(spaceStorage, body.space, next);
        // Only memberspace has a member list, so this also withdraws a
        // published list when a migration discards it.
        if (areMembersPublished(row.policy) !== areMembersPublished(next)) {
          writes.push(
            ...(await memberListWrites(spaceStorage, body.space, next)),
          );
        }
        await spaceStorage.putSpace({ ...row, policy: next });
        if (row.policy?.$type !== next.$type) {
          await spaceStorage.removeAllMembers(body.space);
        }
        await publish(body.space, writes);
        return Response.json({});
      },
    },

    '/xrpc/com.atproto.space.deleteSpace': {
      method: 'POST',
      auth: 'required',
      handler: async (request, _url, auth) => {
        const body = await readJson(request);
        if (!body?.space) {
          return errorResponse('InvalidRequest', 'space is required');
        }
        const scopeError = checkManageScope(auth, body.space, 'delete');
        if (scopeError) return scopeError;

        const row = await ownedSpace(body.space, auth);
        if (row instanceof Response) return row;

        const recipients = await spaceStorage.listCredentialRecipients(
          body.space,
        );
        await spaceStorage.deleteSpace(body.space, new Date().toISOString());
        // Retried until each registered syncer has it or its registration
        // expires. A syncer that still misses it learns from SpaceDeleted on
        // its next credential request.
        await outbox.deliverSpaceDeleted(body.space, recipients);
        return Response.json({});
      },
    },

    // The policy is the owner's own state, so only the owner's session reads
    // it. A policy can publish itself to readers instead.
    '/xrpc/com.atproto.space.getSpace': {
      auth: 'optional',
      handler: async (_request, url, auth) => {
        const space = url.searchParams.get('space');
        if (!space) return errorResponse('InvalidRequest', 'space is required');
        // Whether the space exists is answered before who may read it: a
        // space's address is its owner's DID and a key, so existence is not a
        // secret, and the git helper tells a private repository from a public
        // one by asking this with no session at all.
        const row = await spaceStorage.getSpace(space);
        if (!row?.isOwner || row.deletedAt) {
          return errorResponse('SpaceNotFound', 'Space not found', 404);
        }
        if (!auth) {
          return errorResponse(
            'AuthenticationRequired',
            "The space owner's session is required",
            401,
          );
        }
        const owned = await ownedSpace(space, auth);
        if (owned instanceof Response) return owned;
        return Response.json({ uri: row.uri, policy: row.policy });
      },
    },

    '/xrpc/com.atproto.space.removeWriter': {
      method: 'POST',
      auth: 'required',
      handler: async (request, _url, auth) => {
        const body = await readJson(request);
        if (!body?.space || !body?.did) {
          return errorResponse('InvalidRequest', 'space and did are required');
        }
        const scopeError = checkManageScope(auth, body.space, 'update');
        if (scopeError) return scopeError;
        const row = await ownedSpace(body.space, auth);
        if (row instanceof Response) return row;
        if (row.deletedAt) {
          return errorResponse('SpaceNotFound', 'Space not found', 404);
        }
        // The account is tracked again if it writes while still authorized.
        await spaceStorage.removeSpaceWriter(body.space, body.did);
        return Response.json({});
      },
    },

    '/xrpc/com.atproto.memberspace.putMember': {
      method: 'POST',
      auth: 'required',
      handler: async (request, _url, auth) => {
        const body = await readJson(request);
        if (!body?.space || !body?.did) {
          return errorResponse('InvalidRequest', 'space and did are required');
        }
        if (typeof body.read !== 'boolean' || typeof body.write !== 'boolean') {
          return errorResponse('InvalidRequest', 'read and write are booleans');
        }
        const scopeError = checkManageScope(auth, body.space, 'update');
        if (scopeError) return scopeError;
        const row = await memberspace(body.space, auth);
        if (row instanceof Response) return row;

        const access = { read: body.read, write: body.write };
        await spaceStorage.putMember(body.space, body.did, access);
        if (!access.write) {
          await spaceStorage.removeSpaceWriter(body.space, body.did);
        }
        await publish(
          body.space,
          await memberRecordWrites(
            spaceStorage,
            body.space,
            row.policy,
            body.did,
            access,
          ),
        );
        return Response.json({});
      },
    },

    '/xrpc/com.atproto.memberspace.removeMember': {
      method: 'POST',
      auth: 'required',
      handler: async (request, _url, auth) => {
        const body = await readJson(request);
        if (!body?.space || !body?.did) {
          return errorResponse('InvalidRequest', 'space and did are required');
        }
        const scopeError = checkManageScope(auth, body.space, 'update');
        if (scopeError) return scopeError;
        const row = await memberspace(body.space, auth);
        if (row instanceof Response) return row;
        // Credentials already issued stay valid until they expire.
        await spaceStorage.removeMember(body.space, body.did);
        await spaceStorage.removeSpaceWriter(body.space, body.did);
        await publish(
          body.space,
          await memberRecordWrites(
            spaceStorage,
            body.space,
            row.policy,
            body.did,
          ),
        );
        return Response.json({});
      },
    },

    '/xrpc/com.atproto.memberspace.listMembers': {
      auth: 'required',
      handler: async (_request, url, auth) => {
        const space = url.searchParams.get('space');
        if (!space) return errorResponse('InvalidRequest', 'space is required');

        const raw = url.searchParams.get('limit');
        const limit = raw === null ? 100 : Number(raw);
        if (!Number.isInteger(limit) || limit < 1 || limit > 1000) {
          return errorResponse(
            'InvalidRequest',
            'limit must be an integer between 1 and 1000',
          );
        }

        const row = await memberspace(space, auth);
        if (row instanceof Response) return row;

        const { members, cursor } = await spaceStorage.listMembers(
          space,
          url.searchParams.get('cursor'),
          limit,
        );
        return Response.json({
          ...(cursor ? { cursor } : {}),
          members,
        });
      },
    },

    // A repo host telling this authority that one of its users wrote. Maintains
    // the writer set, which is what listRepos enumerates as the sync boundary.
    '/xrpc/com.atproto.space.notifyWrite': {
      method: 'POST',
      handler: async (request) => {
        const body = await readJson(request);
        const repoRev = body?.repoRev;
        if (
          !body?.space ||
          !body?.repo ||
          !repoRev ||
          typeof body.data?.$bytes !== 'string'
        ) {
          return errorResponse(
            'InvalidRequest',
            'space, repo, repoRev and data are required',
          );
        }
        if (!isTid(repoRev)) {
          return errorResponse('InvalidRequest', 'repoRev must be a TID');
        }

        const hosted = await getDid();
        if (!hosted) {
          return errorResponse('SpaceNotFound', 'Server not initialised', 404);
        }

        /** @type {{iss: string}} */
        let caller;
        try {
          caller = await verifyServiceAuth({
            jwt: (request.headers.get('authorization') ?? '').replace(
              /^Bearer\s+/i,
              '',
            ),
            aud: [spaceHostAudience(hosted), hosted],
            lxm: 'com.atproto.space.notifyWrite',
            resolveDid,
            verifier,
          });
        } catch (err) {
          if (err instanceof SpaceTokenError) {
            return errorResponse(err.code, err.message, 401);
          }
          throw err;
        }

        // A repo host may only report its own account's writes.
        if (caller.iss !== body.repo) {
          return errorResponse(
            'Forbidden',
            'Can only report writes for the calling account',
            403,
          );
        }

        const row = await spaceStorage.getSpace(body.space);
        if (!row?.isOwner || row.deletedAt) {
          return errorResponse('SpaceNotFound', 'Space not found', 404);
        }

        if (isFutureRev(repoRev)) {
          return errorResponse('FutureRev', 'repoRev is too far in the future');
        }

        // The write perimeter, which the authority holds alone. A writer the
        // space does not admit stays out of the writer set, so listRepos never
        // names them and no syncer pulls their repo.
        const authorized = await authorizeSpaceUser({
          spaceStorage,
          spaceRow: row,
          access: 'write',
          userDid: body.repo,
          resolveDid,
          getSigner: ctx.getSigner,
          fetch: ctx.fetch,
        });
        if (!authorized) {
          return errorResponse(
            'Forbidden',
            'Writer is not authorized for this space',
            403,
          );
        }

        /** @type {Uint8Array} */
        let data;
        try {
          data = Uint8Array.from(atob(body.data.$bytes), (c) =>
            c.charCodeAt(0),
          );
        } catch {
          return errorResponse('InvalidRequest', 'data is not base64');
        }
        const sequenced = await recordWriterUpdate(spaceStorage, {
          space: body.space,
          did: body.repo,
          repoRev,
          data,
        });
        // A repeated or out-of-order notice: the writer set already holds this
        // rev or a later one.
        if (!sequenced) return Response.json({});
        // The authority is the only party that knows who is following this
        // space, so forwarding is its job. A writer's own host cannot do it: the
        // registrations are held here.
        await forwardToSyncers(ctx, {
          authorityDid: hosted,
          space: body.space,
          repo: body.repo,
          repoRev,
          data,
          ...sequenced,
        });
        return Response.json({});
      },
    },

    // A space's authority telling this repo host that credentials it issued
    // are revoked before they expire. Kept long enough to outlive any of them.
    '/xrpc/com.atproto.space.notifyCredentialRevoked': {
      method: 'POST',
      handler: async (request) => {
        const body = await readJson(request);
        const credentials = body?.credentials;
        if (
          typeof body?.space !== 'string' ||
          !Array.isArray(credentials) ||
          credentials.length < 1 ||
          credentials.length > 100 ||
          !credentials.every((jti) => typeof jti === 'string' && jti)
        ) {
          return errorResponse(
            'InvalidRequest',
            'space and 1 to 100 credential ids are required',
          );
        }
        /** @type {string} */
        let authority;
        try {
          authority = parseSpaceUri(body.space).spaceDid;
        } catch {
          return errorResponse(
            'InvalidSpaceUri',
            `Not a space uri: ${body.space}`,
          );
        }

        const hosted = await getDid();
        if (!hosted) {
          return errorResponse('Forbidden', 'Server not initialised', 403);
        }
        /** @type {{iss: string}} */
        let caller;
        try {
          caller = await verifyServiceAuth({
            jwt: (request.headers.get('authorization') ?? '').replace(
              /^Bearer\s+/i,
              '',
            ),
            aud: hosted,
            lxm: 'com.atproto.space.notifyCredentialRevoked',
            resolveDid,
            verifier,
          });
        } catch (err) {
          if (err instanceof SpaceTokenError) {
            return errorResponse(err.code, err.message, 401);
          }
          throw err;
        }
        if (caller.iss !== authority) {
          return errorResponse(
            'Forbidden',
            'Only the space authority may revoke its credentials',
            403,
          );
        }

        await spaceStorage.putCredentialRevocations(
          body.space,
          [...new Set(/** @type {string[]} */ (credentials))],
          new Date(Date.now() + REVOCATION_RETENTION_MS).toISOString(),
        );
        return Response.json({});
      },
    },

    // A syncer subscribing to write notifications. Authenticated with a space
    // credential, so only someone the authority already admitted can subscribe.
    '/xrpc/com.atproto.space.registerNotify': {
      method: 'POST',
      handler: async (request) => {
        const body = await readJson(request);
        if (!body?.space || !body?.service) {
          return errorResponse(
            'InvalidRequest',
            'space and service are required',
          );
        }

        const denied = await requireSpaceCredential(request, body.space);
        if (denied) return denied;

        const owned = await hostedSpace(body.space);
        if (owned instanceof Response) return owned;

        // The delivery URL comes from the subscriber's own DID document, so a
        // registration can only ever point at an endpoint that service
        // published for itself.
        const didDoc = await resolveDid(body.service.split('#')[0]);
        const endpoint = didDoc && namedServiceEndpoint(didDoc, body.service);
        if (!endpoint) {
          return errorResponse(
            'ServiceNotResolvable',
            `Could not resolve a service endpoint for ${body.service}`,
          );
        }

        const expiresAt = new Date(Date.now() + REGISTRATION_TTL_MS)
          .toISOString()
          .replace(/\.\d{3}Z$/, '.000Z');
        await spaceStorage.putCredentialRecipient(
          body.space,
          body.service,
          endpoint,
          expiresAt,
        );
        return Response.json({ expiresAt });
      },
    },

    // Withdrawing a registration. The service identifier is not resolved here:
    // a subscriber whose DID document has changed must still be able to leave.
    '/xrpc/com.atproto.space.unregisterNotify': {
      method: 'POST',
      handler: async (request) => {
        const body = await readJson(request);
        if (!body?.space || !body?.service) {
          return errorResponse(
            'InvalidRequest',
            'space and service are required',
          );
        }

        const denied = await requireSpaceCredential(request, body.space);
        if (denied) return denied;

        const owned = await hostedSpace(body.space);
        if (owned instanceof Response) return owned;

        // Idempotent: succeeds whether or not a registration existed.
        await spaceStorage.deleteCredentialRecipient(body.space, body.service);
        return Response.json({});
      },
    },

    // An authority telling us one of its spaces is gone.
    '/xrpc/com.atproto.space.notifySpaceDeleted': {
      method: 'POST',
      handler: async (request) => {
        const body = await readJson(request);
        if (!body?.space) {
          return errorResponse('InvalidRequest', 'space is required');
        }

        const hosted = await getDid();
        if (!hosted) {
          return errorResponse('SpaceNotFound', 'Server not initialised', 404);
        }

        /** @type {{iss: string}} */
        let caller;
        try {
          caller = await verifyServiceAuth({
            jwt: (request.headers.get('authorization') ?? '').replace(
              /^Bearer\s+/i,
              '',
            ),
            aud: hosted,
            lxm: 'com.atproto.space.notifySpaceDeleted',
            resolveDid,
            verifier,
          });
        } catch (err) {
          if (err instanceof SpaceTokenError) {
            return errorResponse(err.code, err.message, 401);
          }
          throw err;
        }

        // Only the space's own authority may declare it deleted.
        const { spaceDid } = parseSpaceUri(body.space);
        if (caller.iss !== spaceDid) {
          return errorResponse(
            'Forbidden',
            'Only the space authority may delete a space',
            403,
          );
        }

        const row = await spaceStorage.getSpace(body.space);
        if (row) {
          // Flag the member's repo rather than erasing it: the records are the
          // member's own, and the proposal has repo hosts keep them.
          await spaceStorage.deleteSpace(body.space, new Date().toISOString());
        }
        return Response.json({});
      },
    },
  };
}