// @pdsjs/core/handlers/xrpc-blob - blob upload, download and listing.
//
// uploadBlob is com.atproto.repo.*, getBlob and listBlobs are
// com.atproto.sync.*. All three read or write blob storage rather than the repo,
// so they are together here.

import { parseByteRange } from '../http.js';
import {
  blobCidFromDigest,
  cidToString,
  isSpaceRecordUri,
  sniffMimeType,
} from '../repo.js';
import { ScopePermissions } from '../scope.js';
import { createSha256 } from '../sha256.js';

/**
 * @typedef {Object} BlobContext
 * @property {import('../ports.js').ActorStoragePort} actorStorage
 * @property {import('../ports.js').BlobPort} blobs
 * @property {number} blobUploadLimit - Largest accepted upload, in bytes
 * @property {import('../ports.js').HasherFactory} [hasher] - SHA-256 factory; absent, the plain JavaScript one
 * @property {boolean} readOnly
 * @property {() => Promise<string|null>} getDid
 * @property {() => Response} readOnlyError
 */

// sniffMimeType reads at most this many bytes of the content.
const SNIFF_LENGTH = 12;

/**
 * A BlobWriter over the one-shot `put`, for a port without `createWriter`.
 * It holds every chunk, so it costs what the buffered upload path always
 * cost; the upload limit bounds it.
 * @param {import('../ports.js').BlobPort} blobs
 * @param {string} did
 * @returns {import('../ports.js').BlobWriter}
 */
function createBufferedWriter(blobs, did) {
  /** @type {Uint8Array[]} */
  let chunks = [];
  return {
    async write(chunk) {
      chunks.push(chunk);
    },
    async commit(cid, mimeType) {
      let size = 0;
      for (const chunk of chunks) size += chunk.length;
      const data = new Uint8Array(size);
      let at = 0;
      for (const chunk of chunks) {
        data.set(chunk, at);
        at += chunk.length;
      }
      await blobs.put(did, cid, data, mimeType);
    },
    async abort() {
      chunks = [];
    },
  };
}

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

/**
 * @param {BlobContext} ctx
 * @returns {{ routes: import('../pds.js').Routes }}
 */
export function createBlobHandlers(ctx) {
  const {
    actorStorage,
    blobs,
    blobUploadLimit,
    hasher = createSha256,
    readOnly,
    getDid,
    readOnlyError,
  } = ctx;

  /**
   * com.atproto.repo.uploadBlob
   * @param {Request} request
   * @param {URL} _url
   * @param {import('../pds.js').Auth|null} auth
   * @returns {Promise<Response>}
   */
  async function handleUploadBlob(request, _url, auth) {
    if (readOnly) return readOnlyError();

    if (!auth) {
      return errorResponse(
        'AuthenticationRequired',
        'Authentication required',
        401,
      );
    }

    const did = await getDid();
    if (!did) {
      return errorResponse('InvalidRequest', 'PDS not initialized');
    }

    const contentType =
      request.headers.get('content-type') || 'application/octet-stream';

    // Deliberately not gated on checkAccountActive: an incoming migration
    // uploads its blobs while the account is still deactivated, before
    // activateAccount. A blob with no record referencing it is inert and gets
    // reaped as an orphan, and record writes remain gated, so accepting blobs
    // in this window is safe and matches the reference PDS.

    const tooLarge = () =>
      errorResponse(
        'PayloadTooLarge',
        `Blob exceeds the maximum size of ${blobUploadLimit} bytes`,
        413,
      );

    const declared = Number(request.headers.get('content-length'));
    if (Number.isFinite(declared) && declared > blobUploadLimit) {
      return tooLarge();
    }

    // The body streams through: each chunk feeds the hash and the writer, and
    // only the sniffing prefix stays in memory. The size check runs per chunk
    // because content-length is not required to be present or honest.
    const hash = hasher();
    const writer = blobs.createWriter
      ? await blobs.createWriter(did)
      : createBufferedWriter(blobs, did);
    let size = 0;
    let head = new Uint8Array(0);
    try {
      if (request.body) {
        const reader = request.body.getReader();
        while (true) {
          const { done, value } = await reader.read();
          if (done) break;
          const chunk =
            value instanceof Uint8Array ? value : new Uint8Array(value);
          size += chunk.length;
          if (size > blobUploadLimit) {
            await reader.cancel();
            await writer.abort();
            return tooLarge();
          }
          if (head.length < SNIFF_LENGTH) {
            const take = Math.min(SNIFF_LENGTH - head.length, chunk.length);
            const grown = new Uint8Array(head.length + take);
            grown.set(head);
            grown.set(chunk.subarray(0, take), head.length);
            head = grown;
          }
          await hash.update(chunk);
          await writer.write(chunk);
        }
      }

      const mimeType = sniffMimeType(head) || contentType;

      const permissions = new ScopePermissions(auth.scope);
      if (!permissions.allowsBlob(mimeType)) {
        await writer.abort();
        return errorResponse(
          'Forbidden',
          `Missing required scope for blob upload of type ${mimeType}`,
          403,
        );
      }

      const cid = cidToString(blobCidFromDigest(await hash.digest()));

      await actorStorage.putBlob(cid, mimeType, size);
      await writer.commit(cid, mimeType);

      return Response.json({
        blob: {
          $type: 'blob',
          ref: { $link: cid },
          mimeType,
          size,
        },
      });
    } catch (e) {
      // Failed reads and failed staging writes end the same way: nothing
      // stays behind under a temporary key.
      try {
        await writer.abort();
      } catch {
        // the original error is the one to surface
      }
      throw e;
    }
  }

  /**
   * The response headers every getBlob answer carries. Disposition and CSP
   * force download and block execution so hostile blobs can't run in the PDS
   * origin; browsers still play such a response inside a video or audio
   * element, since disposition applies to navigation, not embedding.
   * @param {string} cid
   * @returns {Record<string, string>}
   */
  function blobHeaders(cid) {
    return {
      'Cache-Control': 'public, max-age=31536000, immutable',
      'X-Content-Type-Options': 'nosniff',
      'Content-Disposition': `attachment; filename="${cid}"`,
      'Content-Security-Policy': "default-src 'none'; sandbox",
      'Accept-Ranges': 'bytes',
    };
  }

  /**
   * com.atproto.sync.getBlob
   * @param {Request} request
   * @param {URL} url
   * @returns {Promise<Response>}
   */
  async function handleGetBlob(request, url) {
    const did = url.searchParams.get('did');
    const cid = url.searchParams.get('cid');

    if (!did || !cid) {
      return errorResponse('InvalidRequest', 'Missing required parameters');
    }

    const myDid = await getDid();
    if (did !== myDid) {
      return errorResponse('InvalidRequest', 'DID does not match this PDS');
    }

    if (!cid.startsWith('b') || cid.length < 10) {
      return errorResponse('InvalidRequest', 'Invalid CID format');
    }

    // A blob only space records reference is private data. This endpoint is
    // unauthenticated, so such a blob is absent here; com.atproto.space.getBlob
    // serves it to a session or a space credential. A blob with no references
    // at all stays served: the public write flow uploads before it commits.
    const uris = await actorStorage.listBlobRecordUris(cid);
    if (uris.length > 0 && uris.every(isSpaceRecordUri)) {
      return errorResponse('BlobNotFound', 'Blob not found', 404);
    }

    // Range service needs the blob's size before any bytes move, and the
    // metadata row holds it. Without that row (an old mock, or bytes that
    // arrived outside the upload paths) the whole object goes out buffered,
    // as it always did.
    const meta = actorStorage.getBlob ? await actorStorage.getBlob(cid) : null;
    if (!meta) {
      const result = await blobs.get(did, cid);
      if (!result) {
        return errorResponse('BlobNotFound', 'Blob not found', 404);
      }
      return new Response(/** @type {BodyInit} */ (result.data), {
        headers: {
          ...blobHeaders(cid),
          'Content-Type': result.mimeType,
          'Content-Length': String(result.data.length),
        },
      });
    }

    const range = parseByteRange(request.headers.get('range'), meta.size);
    if (range === 'unsatisfiable') {
      return new Response(null, {
        status: 416,
        headers: {
          'Content-Range': `bytes */${meta.size}`,
          'Accept-Ranges': 'bytes',
        },
      });
    }

    /** @type {BodyInit} */
    let body;
    /** @type {string} */
    let mimeType;
    if (blobs.getStream) {
      const found = await blobs.getStream(did, cid, range ?? undefined);
      if (!found) {
        return errorResponse('BlobNotFound', 'Blob not found', 404);
      }
      body = found.body;
      mimeType = found.mimeType;
    } else {
      const result = await blobs.get(did, cid);
      if (!result) {
        return errorResponse('BlobNotFound', 'Blob not found', 404);
      }
      body = /** @type {BodyInit} */ (
        range ? result.data.subarray(range.start, range.end + 1) : result.data
      );
      mimeType = result.mimeType;
    }

    const length = range ? range.end - range.start + 1 : meta.size;
    /** @type {Record<string, string>} */
    const headers = {
      ...blobHeaders(cid),
      'Content-Type': mimeType,
      'Content-Length': String(length),
    };
    if (range) {
      headers['Content-Range'] =
        `bytes ${range.start}-${range.end}/${meta.size}`;
    }
    return new Response(body, { status: range ? 206 : 200, headers });
  }

  /**
   * com.atproto.sync.listBlobs
   * @param {Request} _request
   * @param {URL} url
   * @returns {Promise<Response>}
   */
  async function handleListBlobs(_request, url) {
    const limit = Math.min(
      parseInt(url.searchParams.get('limit') || '500', 10),
      1000,
    );
    const cursor = url.searchParams.get('cursor');

    const result = await actorStorage.listBlobs(cursor, limit);

    // Space-only blobs are absent from the public listing the same way they
    // are absent from getBlob. The cursor advances over the unfiltered page,
    // so a filtered-out blob costs a page slot but never stalls paging.
    /** @type {string[]} */
    const cids = [];
    for (const cid of result.cids) {
      const uris = await actorStorage.listBlobRecordUris(cid);
      if (uris.length > 0 && uris.every(isSpaceRecordUri)) continue;
      cids.push(cid);
    }

    return Response.json({
      cids,
      // Match the reference PDS: omit cursor on a final page rather than
      // emitting `cursor: null`, which strict lexicon validators reject.
      ...(result.cursor ? { cursor: result.cursor } : {}),
    });
  }

  return {
    routes: {
      '/xrpc/com.atproto.repo.uploadBlob': {
        method: 'POST',
        auth: 'required',
        // The Bluesky video service uploads a processed video here with the
        // service auth token the account minted for this server.
        serviceAuth: true,
        handler: handleUploadBlob,
      },
      '/xrpc/com.atproto.sync.getBlob': {
        handler: handleGetBlob,
      },
      '/xrpc/com.atproto.sync.listBlobs': {
        handler: handleListBlobs,
      },
    },
  };
}