// Large bodies: the bytes a record names, on a bus with no PDS under it (design §18, plan 4.4).
//
// Every other record in a private space is small enough to travel inside its own envelope. A picture
// is not: `com.disnetdev.radial.image` carries a `blob` the lexicon caps at ten megabytes, and an
// envelope is capped at half a megabyte because an envelope is the thing a stranger can make a peer
// hold. So bytes travel separately — and the whole question is what makes them trustworthy when they
// arrive without a signature of their own.
//
// **Content addressing is the authentication, and it is end to end.** A blob's name is the CID of its
// bytes; the record that names it is signed inside an envelope; the envelope's `recordCid` covers
// that record. A receiver recomputes the CID of the bytes it was handed and keeps them only under
// that computed name, so a peer cannot substitute one blob for another any more than it can
// substitute one record for another — the substitution simply lands under a different name, which
// nothing references. There is nothing here to trust a peer about, which is why blob transfer needs
// no second signature and must not grow one (ADR §13's argument, applied to bytes).
//
// The CID rule is atproto's, not an invention: CIDv1, **raw** codec, sha2-256 — exactly what
// `com.atproto.repo.uploadBlob` returns for the same bytes. That is what keeps design §18's
// migration obligation true for images too: a private image record is byte-identical to the record a
// public upload would have produced, so republishing a private space when permissioned data lands
// means re-uploading the bytes, not rewriting anybody's records.
//
// One asymmetry with envelopes is worth stating, because it is why there is no blob want list here.
// An evicted envelope leaves no trace — no summary mentions it, so only a durable want can say it was
// ever lost (`quarantine.ts`). A missing blob always leaves a trace: the signed record that names it
// is in the store. **What a replica is missing is therefore derived** (`missingBlobs`), never
// recorded, and so it cannot go stale, cannot be lost by a restart, and cannot disagree with the
// corpus it is computed from.

import { IMAGE_LIMITS } from '../image.js'
import { base32Encode, concat } from './bytes.js'
import { sha256 } from './cid.js'

/** CIDv1, raw (0x55), sha2-256 — the identity atproto gives blob bytes. */
const CID_VERSION = 0x01
const CODEC_RAW = 0x55
const HASH_SHA2_256 = 0x12
const HASH_LENGTH = 32

/**
 * The CID a PDS would return for these bytes, computed locally.
 *
 * Deliberately the same shape as `cidForBytes` in `cid.ts` with one byte changed, and deliberately
 * not shared with it: a record CID is over DAG-CBOR and a blob CID is over the bytes themselves, and
 * a helper that took the codec as a parameter would be one call site away from minting a record CID
 * under the raw codec.
 */
export async function blobCid(bytes: Uint8Array): Promise<string> {
  const digest = await sha256(bytes)
  return `b${base32Encode(concat(new Uint8Array([CID_VERSION, CODEC_RAW, HASH_SHA2_256, HASH_LENGTH]), digest))}`
}

/** The shape a lexicon `blob` field holds — atproto's, so a private record validates unchanged. */
export interface PrivateBlobRef {
  $type: 'blob'
  ref: { $link: string }
  mimeType: string
  size: number
}

/**
 * The blob reference a private writer signs into a record.
 *
 * On the public path this comes back from `uploadBlob`; here the writer mints it, and the bytes go
 * into this replica's blob store rather than into a PDS. Same fields, same CID, same validation.
 */
export async function privateBlobRef(bytes: Uint8Array, mimeType: string): Promise<PrivateBlobRef> {
  return {
    $type: 'blob',
    ref: { $link: await blobCid(bytes) },
    mimeType,
    size: bytes.length,
  }
}

/** A blob named by a record: what to ask a peer for, and what to check when it answers. */
export interface BlobNeed {
  cid: string
  mimeType: string
  size: number
}

const isObject = (value: unknown): value is Record<string, unknown> =>
  typeof value === 'object' && value !== null && !Array.isArray(value)

const isBlobCid = (cid: string): boolean => /^b[a-z2-7]{58}$/.test(cid)

/**
 * Every blob a record value names, found by walking the value rather than by knowing which
 * collections carry one.
 *
 * A lexicon may grow a blob field on any record, and a scan keyed to `image` would go on being right
 * until the day it silently was not — a record would name bytes no replica ever fetched, and the
 * absence would show up as a broken picture rather than as an outstanding want.
 */
export function blobRefsIn(value: unknown, found: BlobNeed[] = []): BlobNeed[] {
  if (Array.isArray(value)) {
    for (const item of value) blobRefsIn(item, found)
    return found
  }
  if (!isObject(value)) return found
  const ref = value.ref
  if (
    value.$type === 'blob' &&
    isObject(ref) &&
    typeof ref.$link === 'string' &&
    typeof value.mimeType === 'string' &&
    typeof value.size === 'number' &&
    isBlobCid(ref.$link)
  ) {
    if (!found.some((need) => need.cid === ref.$link)) {
      found.push({ cid: ref.$link, mimeType: value.mimeType, size: value.size })
    }
    return found
  }
  for (const nested of Object.values(value)) blobRefsIn(nested, found)
  return found
}

export interface BlobLimits {
  /**
   * The largest blob this build will hold.
   *
   * Not a policy dial: a lower value would refuse bytes a record that validates may name, and a
   * higher one would accept bytes no record could ever name. So it is read off the lexicon rather
   * than restated beside it — `image.blob` is the only blob any Radial record carries today, and the
   * day a second one appears with a larger ceiling this has to become the maximum over both.
   */
  maxBlobBytes: number
  /** Bytes of blob payload in one wire frame. Base64 inflates this by a third; keep it well under
   * `MAX_FRAME_BYTES`. */
  chunkBytes: number
}

export const DEFAULT_BLOB_LIMITS: BlobLimits = {
  maxBlobBytes: IMAGE_LIMITS.maxBytes,
  chunkBytes: 256 * 1024,
}

/**
 * Bytes, by their own CID.
 *
 * Content-addressed and therefore append-only and conflict-free: two replicas holding the same name
 * hold the same bytes, or one of them is not holding what it thinks it is — which `put` is what
 * prevents, by computing the name rather than accepting one.
 */
export interface PrivateBlobStore {
  /**
   * Store bytes under their computed CID and return it. Idempotent.
   *
   * `expectedCid` is for the fetch path: bytes assembled from a peer's chunks are checked against
   * the name they were asked for, so a peer that answered with something else fails here rather than
   * quietly filling the store with bytes nothing references.
   */
  put(bytes: Uint8Array, expectedCid?: string): Promise<string>
  get(cid: string): Uint8Array | undefined
  has(cid: string): boolean
  /** Every CID held, sorted — for diagnostics, for export, and for the responder's own bookkeeping. */
  list(): string[]
  close(): void
}

/** The bytes a corpus names and this replica does not hold. Derived, never recorded — see above. */
export function missingBlobs(
  records: readonly { value: unknown }[],
  blobs: Pick<PrivateBlobStore, 'has'>,
): BlobNeed[] {
  const needed: BlobNeed[] = []
  for (const record of records) blobRefsIn(record.value, needed)
  return needed
    .filter((need) => !blobs.has(need.cid))
    .sort((left, right) => (left.cid < right.cid ? -1 : left.cid > right.cid ? 1 : 0))
}

export class MemoryPrivateBlobStore implements PrivateBlobStore {
  readonly #blobs = new Map<string, Uint8Array>()
  readonly #limits: BlobLimits

  constructor(limits: BlobLimits = DEFAULT_BLOB_LIMITS) {
    this.#limits = limits
  }

  async put(bytes: Uint8Array, expectedCid?: string): Promise<string> {
    const cid = await verifyBlobBytes(bytes, expectedCid, this.#limits)
    if (!this.#blobs.has(cid)) this.#blobs.set(cid, bytes.slice())
    return cid
  }

  get(cid: string): Uint8Array | undefined {
    const held = this.#blobs.get(cid)
    return held ? held.slice() : undefined
  }

  has(cid: string): boolean {
    return this.#blobs.has(cid)
  }

  list(): string[] {
    return [...this.#blobs.keys()].sort()
  }

  close(): void {}
}

/**
 * The check every store's `put` runs, in one place because every implementation has to run the same
 * one: bounds first, then the name, then the name it was asked for.
 *
 * Throws rather than returning a reason. Unlike an envelope — where a rejection is one line in a
 * report about somebody else's record — bytes that fail here are always a local write or an answer to
 * something this replica explicitly asked for, and both have a caller waiting.
 */
export async function verifyBlobBytes(
  bytes: Uint8Array,
  expectedCid: string | undefined,
  limits: BlobLimits = DEFAULT_BLOB_LIMITS,
): Promise<string> {
  if (bytes.length === 0) throw new TypeError('a blob has no empty form on the private path')
  if (bytes.length > limits.maxBlobBytes) {
    throw new TypeError(`blob of ${bytes.length} bytes exceeds the ${limits.maxBlobBytes} byte cap`)
  }
  const cid = await blobCid(bytes)
  if (expectedCid !== undefined && cid !== expectedCid) {
    throw new TypeError(`blob bytes hash to ${cid}, not to the requested ${expectedCid}`)
  }
  return cid
}