// @pdsjs/spaces/handlers/write - com.atproto.space write endpoints.
//
// createRecord, putRecord, deleteRecord, applyWrites and clearRecords are thin
// shells over applyWrites() in ../writer.js: they validate the request,
// translate it into write descriptors, and shape the response. All the repo
// logic lives there.
//
// Authorization is two checks: the caller may only write its own repo, and the
// token's `space:` scope must cover the target space, action and collection.

import {
  cborDecodeJson,
  createTid,
  findBlobRefs,
  recordTimeOf,
} from '@pdsjs/core/repo';
import { ScopePermissions } from '@pdsjs/core/scope';
import { createWriteNotifier } from '../notify.js';
import { createNotifyOutbox } from '../outbox.js';
import {
  assertNotPolicyRecord,
  PUBLISHED_COLLECTIONS,
  SpacePolicyError,
} from '../policy.js';
import { parseSpaceUri } from '../uri.js';
import {
  applyWrites,
  MAX_WRITES_PER_COMMIT,
  SpaceWriteError,
} from '../writer.js';

const MAX_JSON_BODY = 512 * 1024;

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

/**
 * Parse a JSON body, rejecting oversized or malformed input.
 * @param {Request} request
 * @returns {Promise<{ok: true, body: any}|{ok: false, response: Response}>}
 */
async function readJson(request) {
  const text = await request.text();
  if (text.length > MAX_JSON_BODY) {
    return {
      ok: false,
      response: errorResponse('InvalidRequest', 'Request body too large', 413),
    };
  }
  try {
    return { ok: true, body: JSON.parse(text) };
  } catch {
    return {
      ok: false,
      response: errorResponse('InvalidRequest', 'Invalid JSON body'),
    };
  }
}

/**
 * Space URIs arrive typed only as `at-uri`, so a malformed one is a request
 * error rather than a 500.
 * @param {string} space
 * @returns {Response|null}
 */
function validateSpaceUri(space) {
  if (typeof space !== 'string') {
    return errorResponse('InvalidRequest', 'space is required');
  }
  try {
    parseSpaceUri(space);
    return null;
  } catch {
    return errorResponse('InvalidSpaceUri', `Not a space uri: ${space}`);
  }
}

/**
 * Assert the token's `space:` scope covers this operation.
 *
 * A grant that names no `collection` should fall back to the space type's
 * declared collections. Resolving a space-type declaration means fetching its
 * lexicon, which this implementation does not do yet — so such a grant
 * currently authorizes no write targets. That fails closed, and a grant that
 * names its collections explicitly works as specified.
 *
 * @param {{did: string, scope?: string}|null} auth
 * @param {string} space
 * @param {{action: string, collection?: string}} op
 * @returns {Response|null}
 */
function checkSpaceScope(auth, space, op) {
  const { spaceDid, spaceType, skey } = parseSpaceUri(space);
  const permissions = new ScopePermissions(auth?.scope);
  const allowed = permissions.allowsSpace({
    spaceType,
    authority: spaceDid,
    skey,
    action: op.action,
    collection: op.collection,
    userDid: auth?.did,
  });
  if (!allowed) {
    return errorResponse(
      'InvalidToken',
      `Scope does not permit ${op.action} in ${space}`,
      403,
    );
  }
  return null;
}

/**
 * A writer may only write its own repo.
 * @param {string} repo
 * @param {string} did
 * @returns {Response|null}
 */
function validateRepo(repo, did) {
  if (repo !== did) {
    return errorResponse(
      'Forbidden',
      'repo must match authenticated user',
      403,
    );
  }
  return null;
}

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

/**
 * @param {string} space
 * @param {string} did
 * @param {string} collection
 * @param {string} rkey
 * @returns {string}
 */
function recordUri(space, did, collection, rkey) {
  return `${space}/${did}/${collection}/${rkey}`;
}

const WRITE_ACTIONS = ['create', 'update', 'delete'];

/**
 * Which of the three operations a write in an applyWrites batch is.
 *
 * The proposal made these a closed union discriminated by `$type` after clients
 * had shipped against an `action` string. Both say the same thing, so both are
 * read; anything else is a write this server cannot place.
 *
 * @param {any} write
 * @returns {'create'|'update'|'delete'|null}
 */
function writeAction(write) {
  const fromType = String(write?.$type ?? '').match(
    /^com\.atproto\.space\.applyWrites#(create|update|delete)$/,
  );
  if (fromType) return /** @type {any} */ (fromType[1]);
  if (WRITE_ACTIONS.includes(write?.action)) return write.action;
  return null;
}

/**
 * @param {Object} ctx
 * @param {import('@pdsjs/core/ports').SpaceStoragePort} ctx.spaceStorage
 * @param {(blobCid: string, recordUri: string, recordTime: number|null) => Promise<void>} ctx.linkBlob
 * @param {(recordUri: string) => Promise<void>} ctx.unlinkBlobs
 * @param {() => Promise<string|null>} ctx.getDid
 * @param {() => Promise<import('../token.js').SpaceSigner>} ctx.getSigner
 * @param {import('@pdsjs/core/ports').DidResolverPort} ctx.resolveDid
 * @param {{collections: string[], check: (write: {space: string|null, collection: string, rkey: string, prevValue: unknown|null, nextValue: unknown|null}) => Promise<void>}} [ctx.recordGuard]
 * @param {typeof fetch} [ctx.fetch]
 * @param {(work: Promise<unknown>) => void} [ctx.background] - runs the
 *   syncers' notices after the response; absent, they run before it
 * @param {ReturnType<typeof createNotifyOutbox>} [ctx.outbox] - built here
 *   from the rest of ctx when absent
 * @returns {import('@pdsjs/core/pds').Routes}
 */
export function createWriteRoutes(ctx) {
  const { spaceStorage, linkBlob, unlinkBlobs, getDid, recordGuard } = ctx;
  const outbox = ctx.outbox ?? createNotifyOutbox(ctx);

  /**
   * Consult the write guard for one write in a collection it covers. Null to
   * proceed, the refusal Response otherwise. Read-then-check races the write
   * queue; the single account that writes here makes that a non-event.
   * @param {string} space
   * @param {import('../writer.js').SpaceWriteInput} write
   * @returns {Promise<Response|null>}
   */
  async function guardWrite(space, write) {
    if (!recordGuard || !recordGuard.collections.includes(write.collection)) {
      return null;
    }
    const prev = await spaceStorage.getSpaceRecord(
      space,
      write.collection,
      write.rkey,
    );
    try {
      await recordGuard.check({
        space,
        collection: write.collection,
        rkey: write.rkey,
        prevValue: prev ? cborDecodeJson(prev.value) : null,
        nextValue: write.action === 'delete' ? null : (write.record ?? null),
      });
      return null;
    } catch (err) {
      const message = err instanceof Error ? err.message : String(err);
      const code =
        err instanceof Error && 'code' in err
          ? String(/** @type {{code: unknown}} */ (err).code)
          : 'InvalidRequest';
      return errorResponse(code, message);
    }
  }

  /**
   * Record which blobs a space write leaves referenced.
   *
   * The links go in the account's one blob link table, beside those of public
   * records. Orphan reaping asks that table whether anything still points at a
   * blob, so a space record protects its blobs the same way a public record
   * does, with no second table for the reaper to be taught about.
   *
   * @param {string} space
   * @param {string} did
   * @param {import('../writer.js').SpaceWriteInput[]} writes
   * @returns {Promise<void>}
   */
  async function indexBlobs(space, did, writes) {
    for (const write of writes) {
      const uri = recordUri(space, did, write.collection, write.rkey);
      // An update replaces the record's references rather than adding to them.
      await unlinkBlobs(uri);
      if (write.action === 'delete') continue;
      for (const cid of findBlobRefs(write.record)) {
        await linkBlob(cid, uri, recordTimeOf(write.record));
      }
    }
  }

  const fireNotifyWrite = createWriteNotifier(ctx, outbox);

  /**
   * Shared shell for the three single-write endpoints.
   * @param {Request} request
   * @param {{did: string, scope?: string}|null} auth
   * @param {(body: any) => {action: 'create'|'put'|'delete', collection: string, rkey: string, record?: any}} toWrite
   * @param {(body: any, result: import('../writer.js').SpaceWriteResult, did: string) => any} toBody
   * @returns {Promise<Response>}
   */
  async function singleWrite(request, auth, toWrite, toBody) {
    const parsed = await readJson(request);
    if (!parsed.ok) return parsed.response;
    const body = parsed.body;
    const did = /** @type {{did: string}} */ (auth).did;

    const uriError = validateSpaceUri(body.space);
    if (uriError) return uriError;
    const repoError = validateRepo(body.repo, did);
    if (repoError) return repoError;
    if (typeof body.collection !== 'string') {
      return errorResponse('InvalidRequest', 'collection is required');
    }

    let write;
    try {
      write = toWrite(body);
      assertNotPolicyRecord(write.collection);
    } catch (err) {
      return writeErrorResponse(err);
    }

    // A put may create or update, so resolve which before checking scope, the
    // way the reference implementation does.
    /** @type {string} */
    let action = write.action;
    if (action === 'put') {
      const existing = await spaceStorage.getSpaceRecordCid(
        body.space,
        write.collection,
        write.rkey,
      );
      action = existing ? 'update' : 'create';
    }
    const scopeError = checkSpaceScope(auth, body.space, {
      action,
      collection: write.collection,
    });
    if (scopeError) return scopeError;

    const refused = await guardWrite(body.space, write);
    if (refused) return refused;

    try {
      const commit = await applyWrites(spaceStorage, {
        space: body.space,
        writes: [write],
        owner: did,
      });
      // After the commit: the write is what makes the references real, and a
      // link to a record that failed to write would keep a blob alive forever.
      await indexBlobs(body.space, did, [
        { ...write, rkey: commit.results[0].rkey },
      ]);
      await fireNotifyWrite(body.space, commit);
      return Response.json(toBody(body, commit.results[0], did));
    } catch (err) {
      return writeErrorResponse(err);
    }
  }

  return {
    '/xrpc/com.atproto.space.createRecord': {
      method: 'POST',
      auth: 'required',
      handler: (request, _url, auth) =>
        singleWrite(
          request,
          auth,
          (body) => {
            if (body.record === undefined) {
              throw new SpaceWriteError('record is required', 'InvalidRequest');
            }
            return {
              action: 'create',
              collection: body.collection,
              rkey: body.rkey ?? createTid(),
              record: body.record,
            };
          },
          (body, result, did) => ({
            uri: recordUri(body.space, did, result.collection, result.rkey),
            cid: result.cid,
          }),
        ),
    },

    '/xrpc/com.atproto.space.putRecord': {
      method: 'POST',
      auth: 'required',
      handler: (request, _url, auth) =>
        singleWrite(
          request,
          auth,
          (body) => {
            if (typeof body.rkey !== 'string') {
              throw new SpaceWriteError('rkey is required', 'InvalidRequest');
            }
            if (body.record === undefined) {
              throw new SpaceWriteError('record is required', 'InvalidRequest');
            }
            if (
              body.swapRecord !== undefined &&
              body.swapRecord !== null &&
              typeof body.swapRecord !== 'string'
            ) {
              throw new SpaceWriteError(
                'swapRecord must be a CID string or null',
                'InvalidRequest',
              );
            }
            return {
              action: 'put',
              collection: body.collection,
              rkey: body.rkey,
              record: body.record,
              // A pds.js extension the draft proposal does not have: null
              // requires that the record does not exist, a CID requires that
              // exact version, absent skips the check. A server without it
              // ignores the field, so a client falls back to last-write-wins.
              ...(body.swapRecord !== undefined
                ? { swapCid: body.swapRecord }
                : {}),
            };
          },
          (body, result, did) => ({
            uri: recordUri(body.space, did, result.collection, result.rkey),
            cid: result.cid,
          }),
        ),
    },

    '/xrpc/com.atproto.space.deleteRecord': {
      method: 'POST',
      auth: 'required',
      handler: (request, _url, auth) =>
        singleWrite(
          request,
          auth,
          (body) => {
            if (typeof body.rkey !== 'string') {
              throw new SpaceWriteError('rkey is required', 'InvalidRequest');
            }
            return {
              action: 'delete',
              collection: body.collection,
              rkey: body.rkey,
            };
          },
          () => ({}),
        ),
    },

    '/xrpc/com.atproto.space.applyWrites': {
      method: 'POST',
      auth: 'required',
      handler: async (request, _url, auth) => {
        const parsed = await readJson(request);
        if (!parsed.ok) return parsed.response;
        const body = parsed.body;
        const did = /** @type {{did: string}} */ (auth).did;

        const uriError = validateSpaceUri(body.space);
        if (uriError) return uriError;
        const repoError = validateRepo(body.repo, did);
        if (repoError) return repoError;
        if (!Array.isArray(body.writes)) {
          return errorResponse('InvalidRequest', 'writes must be an array');
        }

        /** @type {import('../writer.js').SpaceWriteInput[]} */
        const writes = [];
        for (const w of body.writes) {
          const collection = w?.collection;
          if (typeof collection !== 'string') {
            return errorResponse(
              'InvalidRequest',
              'each write needs a collection',
            );
          }
          try {
            assertNotPolicyRecord(collection);
          } catch (err) {
            return writeErrorResponse(err);
          }
          const action = writeAction(w);
          if (!action) {
            return errorResponse(
              'InvalidRequest',
              `Unsupported write type: ${w?.$type ?? w?.action ?? '(none)'}`,
            );
          }
          if (action === 'delete') {
            writes.push({ action, collection, rkey: w.rkey });
          } else {
            writes.push({
              action,
              collection,
              // A create may leave the server to mint the key. An empty string
              // is how the earlier clients say so.
              rkey: action === 'create' ? w.rkey || createTid() : w.rkey,
              record: w.value,
            });
          }
        }

        for (const w of writes) {
          const scopeError = checkSpaceScope(auth, body.space, {
            action: w.action,
            collection: w.collection,
          });
          if (scopeError) return scopeError;
        }
        for (const w of writes) {
          const refused = await guardWrite(body.space, w);
          if (refused) return refused;
        }

        try {
          const commit = await applyWrites(spaceStorage, {
            space: body.space,
            writes,
            owner: did,
          });
          await indexBlobs(
            body.space,
            did,
            writes.map((w, i) => ({ ...w, rkey: commit.results[i].rkey })),
          );
          await fireNotifyWrite(body.space, commit);
          return Response.json({
            results: commit.results.map((r) =>
              r.action === 'delete'
                ? { $type: 'com.atproto.space.applyWrites#deleteResult' }
                : {
                    $type: `com.atproto.space.applyWrites#${r.action}Result`,
                    uri: recordUri(body.space, did, r.collection, r.rkey),
                    cid: r.cid,
                  },
            ),
          });
        } catch (err) {
          return writeErrorResponse(err);
        }
      },
    },

    '/xrpc/com.atproto.space.clearRecords': {
      method: 'POST',
      auth: 'required',
      handler: async (request, _url, auth) => {
        const parsed = await readJson(request);
        if (!parsed.ok) return parsed.response;
        const body = parsed.body;
        const did = /** @type {{did: string}} */ (auth).did;

        const uriError = validateSpaceUri(body.space);
        if (uriError) return uriError;
        const repoError = validateRepo(body.repo, did);
        if (repoError) return repoError;

        // In the authority's own repo, the published policy and member
        // records are the PDS's copy of the space's configuration, so they
        // stay, and keeping them takes no grant.
        const keep =
          parseSpaceUri(body.space).spaceDid === did
            ? PUBLISHED_COLLECTIONS
            : [];
        const records = (
          await spaceStorage.listAllSpaceRecords(body.space)
        ).filter((r) => !keep.includes(r.collection));

        for (const collection of new Set(records.map((r) => r.collection))) {
          const scopeError = checkSpaceScope(auth, body.space, {
            action: 'delete',
            collection,
          });
          if (scopeError) return scopeError;
        }
        /** @type {import('../writer.js').SpaceWriteInput[]} */
        const writes = records.map((r) => ({
          action: 'delete',
          collection: r.collection,
          rkey: r.rkey,
        }));
        for (const w of writes) {
          const refused = await guardWrite(body.space, w);
          if (refused) return refused;
        }

        try {
          /** @type {{rev: string, setHash: Uint8Array}|null} */
          let commit = null;
          for (let i = 0; i < writes.length; i += MAX_WRITES_PER_COMMIT) {
            const chunk = writes.slice(i, i + MAX_WRITES_PER_COMMIT);
            commit = await applyWrites(spaceStorage, {
              space: body.space,
              writes: chunk,
              owner: did,
            });
            await indexBlobs(body.space, did, chunk);
          }
          if (commit) await fireNotifyWrite(body.space, commit);
          return Response.json({});
        } catch (err) {
          return writeErrorResponse(err);
        }
      },
    },

    '/xrpc/com.atproto.space.listSpaces': {
      auth: 'required',
      handler: async (_request, url) => {
        const params = url.searchParams;
        const limitParam = params.get('limit');
        let limit = limitParam === null ? 50 : Number(limitParam);
        if (!Number.isInteger(limit) || limit < 1 || limit > 100) {
          if (limitParam !== null) {
            return errorResponse(
              'InvalidRequest',
              'limit must be an integer between 1 and 100',
            );
          }
          limit = 50;
        }

        const { spaces, cursor } = await spaceStorage.listSpaces({
          type: params.get('spaceType') ?? params.get('type'),
          did: params.get('did'),
          cursor: params.get('cursor'),
          limit,
          excludeDeleted: true,
        });

        return Response.json({
          ...(cursor ? { cursor } : {}),
          spaces: spaces.map((s) => ({ uri: s.uri, isOwner: s.isOwner })),
        });
      },
    },
  };
}