/**
 * Content-addressed blob store (SPEC.md §5.9.2).
 *
 * Blobs are the file-attachment bytes a row references through a `blob_ref`
 * column (§2.4 tag 7). Unlike segments (§5.1 TTL cache entries), blobs are
 * **durable**: a blob referenced by a live row must stay downloadable
 * indefinitely (§5.9.5 B3). The store is keyed by `blobId`
 * (`"sha256:" + hex`), shares the S3/R2 backend and presign machinery with
 * segments, and exposes an orphan sweep (§5.9.2) rather than row-tied
 * deletion.
 */
import { emitEvent, type SyncularServerEvents } from './events';
import { sha256Hex } from './scopes';
import type { ServerStorage } from './storage';

export interface BlobRecord {
  readonly blobId: string;
  readonly partition: string;
  readonly byteLength: number;
  /** Advisory MIME type persisted from the upload, if any (§5.9.5). */
  readonly mediaType?: string;
  readonly createdAtMs: number;
}

/**
 * Coarse blob-store counters for the admin/console read surface (work item
 * §2.5), partition-scoped. Optional. `count`/`bytes` are total stored
 * blobs; a store that omits `stats()` cannot report them.
 */
export interface BlobStoreStats {
  readonly count: number;
  readonly bytes: number;
  /**
   * ADDITIVE marker — present and `true` only on stores whose counters are
   * approximate (a LIST-free S3/R2 accumulator). Absent on the exact
   * in-process stores (memory/sqlite). The admin surface carries it through.
   */
  readonly approximate?: boolean;
}

export interface BlobStore {
  /**
   * Store blob bytes under `blobId` (idempotent — same id ⇒ no-op success,
   * §5.9.3). The caller has already verified the content address matches.
   */
  put(
    partition: string,
    blobId: string,
    bytes: Uint8Array,
    nowMs: number,
    mediaType?: string,
  ): Promise<BlobRecord>;
  /** Presence check for the push existence rule (§5.9.6) and download. */
  has(partition: string, blobId: string): Promise<boolean>;
  get(
    partition: string,
    blobId: string,
  ): Promise<{ record: BlobRecord; bytes: Uint8Array } | undefined>;
  /**
   * §5.9.2 orphan sweep: delete blobs uploaded before `olderThanMs` that
   * no live row references. The set of referenced blobIds is supplied by
   * the storage reference index — the store never scans rows itself.
   * Returns the deleted blobIds.
   */
  sweepOrphans(
    partition: string,
    olderThanMs: number,
    referencedBlobIds: ReadonlySet<string>,
  ): Promise<string[]>;
  /** Admin/console counters (work item §2.5) — ADDITIVE, optional. */
  stats?(partition: string): Promise<BlobStoreStats>;
}

/** `"sha256:" + hex` of the bytes — the content address (§5.9.1). */
export async function blobIdFor(bytes: Uint8Array): Promise<string> {
  return `sha256:${await sha256Hex(bytes)}`;
}

const BLOB_ID_PATTERN = /^sha256:[0-9a-f]{64}$/;

export function isBlobId(value: string): boolean {
  return BLOB_ID_PATTERN.test(value);
}

export class MemoryBlobStore implements BlobStore {
  #entries = new Map<string, { record: BlobRecord; bytes: Uint8Array }>();

  #key(partition: string, blobId: string): string {
    return `${partition} ${blobId}`;
  }

  async put(
    partition: string,
    blobId: string,
    bytes: Uint8Array,
    nowMs: number,
    mediaType?: string,
  ): Promise<BlobRecord> {
    const key = this.#key(partition, blobId);
    const existing = this.#entries.get(key);
    if (existing !== undefined) return existing.record;
    const record: BlobRecord = {
      blobId,
      partition,
      byteLength: bytes.length,
      ...(mediaType !== undefined ? { mediaType } : {}),
      createdAtMs: nowMs,
    };
    this.#entries.set(key, { record, bytes: bytes.slice() });
    return record;
  }

  async has(partition: string, blobId: string): Promise<boolean> {
    return this.#entries.has(this.#key(partition, blobId));
  }

  async get(
    partition: string,
    blobId: string,
  ): Promise<{ record: BlobRecord; bytes: Uint8Array } | undefined> {
    return this.#entries.get(this.#key(partition, blobId));
  }

  async sweepOrphans(
    partition: string,
    olderThanMs: number,
    referencedBlobIds: ReadonlySet<string>,
  ): Promise<string[]> {
    const swept: string[] = [];
    for (const [key, { record }] of this.#entries) {
      if (
        record.partition === partition &&
        record.createdAtMs < olderThanMs &&
        !referencedBlobIds.has(record.blobId)
      ) {
        this.#entries.delete(key);
        swept.push(record.blobId);
      }
    }
    return swept;
  }

  async stats(partition: string): Promise<BlobStoreStats> {
    let count = 0;
    let bytes = 0;
    for (const { record } of this.#entries.values()) {
      if (record.partition === partition) {
        count += 1;
        bytes += record.byteLength;
      }
    }
    return { count, bytes };
  }
}

export interface SweepOrphanBlobsOptions {
  /**
   * Grace period: a blob uploaded within the last `graceMs` is NEVER swept,
   * even if unreferenced. This is the whole protection for the §5.9.2
   * upload-before-reference race — a fresh upload is legitimately
   * unreferenced until its push lands, so the grace MUST comfortably exceed
   * any sane upload→push latency. Default 24 h (well above any push window).
   */
  readonly graceMs?: number;
  /** Sweep clock (epoch ms); defaults to `Date.now`. */
  readonly nowMs?: number;
  /** Optional structured-events sink (`blob.swept`). */
  readonly events?: SyncularServerEvents;
}

export interface SweepOrphanBlobsResult {
  readonly partition: string;
  readonly swept: string[];
  readonly referencedCount: number;
}

const DEFAULT_SWEEP_GRACE_MS = 24 * 60 * 60 * 1000;

/**
 * §5.9.2 orphan sweep — the ready-made GC helper. Reads the live keep-set
 * (`storage.listReferencedBlobIds`) and hands it, with the grace cutoff, to
 * `blobStore.sweepOrphans`, which deletes only blobs BOTH unreferenced AND
 * older than the grace period. Returns the swept ids and emits one
 * `blob.swept` ops event.
 *
 * The upload-before-reference race, stated honestly: a client uploads bytes
 * before pushing the row that references them (§5.9.2), so a just-uploaded
 * blob is unreferenced until its push arrives. The grace period is the ONLY
 * thing standing between that fresh blob and deletion — keep it far above any
 * push latency (the 24 h default is deliberately generous). The store's
 * `createdAtMs` (object metadata) is the upload time the cutoff compares
 * against.
 *
 * Requires `storage.listReferencedBlobIds` (the §5.9.4 reference index). A
 * storage backend without it cannot be swept safely — this throws rather
 * than sweeping against an empty keep-set (which would delete everything).
 */
export async function sweepOrphanBlobs(
  storage: ServerStorage,
  blobStore: BlobStore,
  partition: string,
  options?: SweepOrphanBlobsOptions,
): Promise<SweepOrphanBlobsResult> {
  const listReferenced = storage.listReferencedBlobIds;
  if (listReferenced === undefined) {
    throw new Error(
      'sweepOrphanBlobs: storage has no blob reference index ' +
        '(listReferencedBlobIds); refusing to sweep against an empty keep-set',
    );
  }
  const graceMs = options?.graceMs ?? DEFAULT_SWEEP_GRACE_MS;
  const nowMs = options?.nowMs ?? Date.now();
  const olderThanMs = nowMs - graceMs;
  const referenced = new Set(await listReferenced.call(storage, partition));
  const swept = await blobStore.sweepOrphans(
    partition,
    olderThanMs,
    referenced,
  );
  if (options?.events !== undefined) {
    emitEvent(options.events, {
      type: 'blob.swept',
      atMs: nowMs,
      partition,
      swept: swept.length,
      referenced: referenced.size,
      graceMs,
    });
  }
  return { partition, swept, referencedCount: referenced.size };
}