import { Inject, Injectable } from "@nestjs/common";
import { Prisma, type BlockType as PrismaBlockType } from "@prisma/client";

/** JSON column input type (the `content` / `properties` Json fields). */
type PrismaJsonValue = Prisma.InputJsonValue;
import { PrismaService } from "../../common/prisma.service.js";
import { AuditService } from "../../common/audit.service.js";
import { EventBus } from "../../common/event-bus.js";
import { ApiError } from "../../common/errors.js";
import { EVENTS } from "../../constants.js";
import {
  type InternalBlock,
  type InternalFlushInput,
  type InternalReadQueryInput
} from "./internal.dto.js";

/**
 * Internal block endpoints (TASK-0092 slice 2).
 *
 * The service-to-service surface the realtime gateway uses to (a) hydrate a
 * cold room (`internalRead`) and (b) persist — flush — a CRDT doc under the
 * `pages.version` CAS (`internalFlush`). These are the **internal twin** of
 * the public block endpoints: same tables, service-token auth, no
 * permission re-check (the room's join matrix already gated the edits —
 * the flush is a system write, attributed `actor: 'realtime-flush'`).
 *
 * They live behind `InternalServiceGuard` (shared `INTERNAL_SERVICE_TOKEN`,
 * fail-closed) and are EXCLUDED from the global `api/v1` prefix in
 * `main.ts` — the documented contract: `/internal/blocks/*`, never on the
 * public route table.
 *
 * Block-type translation: the codec/DTO `type` is lowercase snake_case
 * (PRODUCT-SPEC §4), the DB enum SCREAMING_SNAKE — the 1:1 case fold at
 * this boundary (same resolution as the realtime `PrismaDocStore` seam).
 */

/** The codec block type (lowercase) → the Prisma enum (uppercase). */
function codecToDbType(t: string): PrismaBlockType {
  return t.toUpperCase() as PrismaBlockType;
}

/** The Prisma enum (uppercase) → the codec block type (lowercase). */
function dbToCodecType(t: PrismaBlockType | string): string {
  return t.toLowerCase();
}

/** Canonical JSON equality (key-sorted stringify) — the blockDelta's
 *  content/properties comparison, the codec `diffBlockSets` semantics. */
function jsonEqual(a: unknown, b: unknown): boolean {
  return canonicalJson(a) === canonicalJson(b);
}

function canonicalJson(value: unknown): string {
  if (value === null || typeof value !== "object") return JSON.stringify(value) ?? "null";
  if (Array.isArray(value)) return `[${value.map((v) => canonicalJson(v)).join(",")}]`;
  const entries = Object.entries(value as Record<string, unknown>)
    .filter(([, v]) => v !== undefined)
    .sort(([ka], [kb]) => (ka < kb ? -1 : ka > kb ? 1 : 0));
  return `{${entries.map(([k, v]) => `${JSON.stringify(k)}:${canonicalJson(v)}`).join(",")}}`;
}

interface BlockRow {
  id: string;
  parentId: string | null;
  type: PrismaBlockType | string;
  position: string;
  content: PrismaJsonValue | null;
  properties: PrismaJsonValue | null;
}

/** The wire block shape (codec convention) for read responses / conflicts. */
interface WireBlock {
  id: string;
  parentId: string | null;
  type: string;
  position: string;
  content: Record<string, unknown>;
  properties?: Record<string, unknown>;
}

function toWireBlock(row: BlockRow): WireBlock {
  const properties = row.properties as Record<string, unknown> | null;
  return {
    id: row.id,
    parentId: row.parentId,
    type: dbToCodecType(row.type),
    position: row.position,
    content: (row.content ?? {}) as Record<string, unknown>,
    ...(properties !== null && properties !== undefined && Object.keys(properties).length > 0
      ? { properties }
      : {})
  };
}

@Injectable()
export class InternalBlocksService {
  // Plain parameter injection — matches PageService (the sibling that boots).
  // The earlier explicit `@Inject(PrismaService)` token failed at boot
  // ("Nest can't resolve dependencies of the InternalBlocksService (?…)")
  // while the identical plain injection in PageService resolved fine, so this
  // class now uses the same proven path. The unit test instantiates the ctor
  // positionally, so it is unaffected. (TASK-0100 ledger: pre-existing API
  // boot regression recorded in TASK-0095 S9; see also TestSeamService,
  // which shares the root-level PrismaService/AuditService providers.)
  constructor(
    @Inject(PrismaService) private readonly prisma: PrismaService,
    @Inject(AuditService) private readonly audit: AuditService,
    @Inject(EventBus) private readonly events: EventBus
  ) {}

  /**
   * `GET /internal/blocks/:pageId` — the hydrator's read. Position-ordered
   * blocks + the `pages.version` (the flush's CAS base), cursor-paginated
   * by `position` (the 100k-block page streams per page — doc §4.2).
   */
  async internalRead(pageId: string, query: InternalReadQueryInput): Promise<{
    pageId: string;
    version: number;
    blocks: WireBlock[];
    nextCursor: string | null;
    hasMore: boolean;
  }> {
    const pageSize = query.pageSize ?? 500;
    const page = await this.prisma.page.findUnique({
      where: { id: pageId },
      select: { workspaceId: true, version: true, deletedAt: true }
    });
    if (page === null || page.deletedAt !== null) {
      throw ApiError.notFound(`Page ${pageId} not found.`);
    }
    const where: Prisma.BlockWhereInput = {
      pageId,
      deletedAt: null,
      ...(query.cursor ? { position: { gt: query.cursor } } : {})
    };
    const rows = await this.prisma.block.findMany({
      where,
      select: { id: true, parentId: true, type: true, position: true, content: true, properties: true },
      orderBy: { position: "asc" },
      take: pageSize + 1 // +1: the hasMore probe (no COUNT query)
    });
    const hasMore = rows.length > pageSize;
    const pageRows = hasMore ? rows.slice(0, pageSize) : rows;
    return {
      pageId,
      version: page.version,
      blocks: pageRows.map(toWireBlock),
      nextCursor: hasMore ? pageRows[pageRows.length - 1].position : null,
      hasMore
    };
  }

  /**
   * `POST /internal/blocks/flush` — the persister's write (doc §4.3).
   *
   * The full block set of the page (the CRDT doc re-encoded) persisted
   * under the `pages.version` CAS:
   *  - `baseVersion` === current version → the version bumps, the set is
   *    upserted (removed rows soft-deleted), `page.updated` is emitted
   *    (the `version` + `blockDelta` additive fields), the flush is
   *    audited `actor: 'realtime-flush'`;
   *  - otherwise a concurrent non-Yjs writer (REST batch) won — the flush
   *    is a NO-OP (nothing is written) and the response carries the
   *    store's current state as `conflict.externalBlocks` for the room to
   *    rebase on (the "flush wins" rule, the 3-way merge in
   *    `apps/realtime/src/rooms.ts`).
   */
  async internalFlush(input: InternalFlushInput): Promise<
    | { version: number }
    | { version: number; conflict: { externalBlocks: WireBlock[] } }
  > {
    const { pageId, baseVersion, blocks } = input;
    const page = await this.prisma.page.findUnique({
      where: { id: pageId },
      select: { workspaceId: true, version: true, deletedAt: true }
    });
    if (page === null || page.deletedAt !== null) {
      throw ApiError.notFound(`Page ${pageId} not found.`);
    }
    if (page.version !== baseVersion) {
      // Fast path: the conflict is already visible before the transaction
      // (the CAS below is the authoritative guard).
      return {
        version: page.version,
        conflict: { externalBlocks: await this.readLiveBlocks(pageId) }
      };
    }
    // The flush's minimal-diff basis: the state AT baseVersion (valid —
    // the version equality means no writer landed since baseVersion).
    const previous = await this.readLiveBlocks(pageId);

    try {
      const newVersion = await this.prisma.$transaction(async (tx) => {
        // The atomic CAS guard (first statement — a racy concurrent writer
        // aborts the whole transaction, nothing is written).
        const cas = await tx.page.updateMany({
          where: { id: pageId, version: baseVersion },
          data: { version: baseVersion + 1, updatedAt: new Date() }
        });
        if (cas.count !== 1) throw new FlushCasConflict();

        const existing = await tx.block.findMany({
          where: { pageId },
          select: { id: true, deletedAt: true }
        });
        const liveIds = new Set(blocks.map((b) => b.id));
        const removed = existing
          .filter((b) => b.deletedAt === null && !liveIds.has(b.id))
          .map((b) => b.id);
        if (removed.length > 0) {
          await tx.block.updateMany({
            where: { id: { in: removed }, deletedAt: null },
            data: { deletedAt: new Date() }
          });
        }
        // TASK-0135 N+1 fix (hot path #2 — the editor autosave flush, the
        // profiled per-block `upsert` loop). All blocks are now upserted in
        // ONE multi-row `INSERT … ON CONFLICT (id) DO UPDATE`, so the flush
        // transaction's query count is CONSTANT (1 CAS + 1 read + ≤1
        // soft-delete + 1 upsert) regardless of block count — flat across 10/
        // 100/1000 blocks (asserted by the query-shape test). Semantics are
        // identical to the per-block upsert: an absent id is created, a
        // present id has its fields updated, a soft-deleted row is REVIVED
        // (`deletedAt = NULL`). `updatedAt` is set explicitly — Prisma's
        // `@updatedAt` does not apply to raw SQL, and the column has no DB
        // default (NOT NULL).
        const now = new Date();
        const upsertRows = blocks.map(
          (b) =>
            Prisma.sql`(
              ${b.id}, ${pageId}, ${page.workspaceId}, ${b.parentId ?? null},
              ${codecToDbType(b.type)}::"BlockType", ${b.position},
              ${JSON.stringify(b.content)}::jsonb, ${JSON.stringify(b.properties ?? {})}::jsonb, ${now}
            )`
        );
        if (upsertRows.length > 0) {
          await tx.$executeRaw`
            INSERT INTO "blocks"
              ("id", "pageId", "workspaceId", "parentId", "type", "position", "content", "properties", "updatedAt")
            VALUES ${Prisma.join(upsertRows, ", ")}
            ON CONFLICT ("id") DO UPDATE SET
              "parentId" = EXCLUDED."parentId",
              "type" = EXCLUDED."type",
              "position" = EXCLUDED."position",
              "content" = EXCLUDED."content",
              "properties" = EXCLUDED."properties",
              "deletedAt" = NULL,
              "updatedAt" = ${now}
          `;
        }
        return baseVersion + 1;
      });

      const blockDelta = this.diffCounts(previous, blocks);
      // Audit: the flush is a system write — attributed, not impersonable
      // (the public surface has no path to set this actor; the service
      // token is env-only).
      await this.audit.write({
        action: "block.flush",
        resourceType: "block",
        resourceId: pageId,
        workspaceId: page.workspaceId,
        pageId,
        metadata: { actor: "realtime-flush", version: newVersion, blocks: blocks.length, ...blockDelta }
      });
      // The Phase-3 event, the `version` + `blockDelta` additive fields
      // (packages/shared PageUpdatedPayload) — the Phase-8 indexer + the
      // computed-recompute consumer both listen.
      await this.events.publish(EVENTS.PAGE_UPDATED, {
        workspaceId: page.workspaceId,
        pageId,
        version: newVersion,
        blockDelta
      });
      return { version: newVersion };
    } catch (err) {
      if (err instanceof FlushCasConflict) {
        // The racy CAS: read back the concurrent writer's state for the
        // room's rebase (the fast path above misses this window).
        const version = await this.prisma.page.findUnique({
          where: { id: pageId },
          select: { version: true }
        });
        return {
          version: version?.version ?? baseVersion + 1,
          conflict: { externalBlocks: await this.readLiveBlocks(pageId) }
        };
      }
      throw err;
    }
  }

  /** The page's current live blocks (position-ordered, codec shape). */
  private async readLiveBlocks(pageId: string): Promise<WireBlock[]> {
    const rows = await this.prisma.block.findMany({
      where: { pageId, deletedAt: null },
      select: { id: true, parentId: true, type: true, position: true, content: true, properties: true },
      orderBy: { position: "asc" }
    });
    return rows.map(toWireBlock);
  }

  /** The flush's minimal op counts vs the last persisted state — the
   *  codec `diffBlockSets` semantics (insert/update/move/delete counts). */
  private diffCounts(
    previous: WireBlock[],
    next: readonly InternalBlock[]
  ): { inserted: number; updated: number; moved: number; deleted: number } {
    const prevById = new Map(previous.map((b) => [b.id, b] as const));
    const nextById = new Map(next.map((b) => [b.id, b] as const));
    let inserted = 0;
    let updated = 0;
    let moved = 0;
    let deleted = 0;
    for (const nb of next) {
      const pb = prevById.get(nb.id);
      if (pb === undefined) {
        inserted += 1;
        continue;
      }
      const contentChanged = !jsonEqual(pb.content, nb.content);
      const prevProps = pb.properties ?? {};
      const nextProps = nb.properties ?? {};
      const propertiesChanged = !jsonEqual(prevProps, nextProps);
      if (contentChanged || propertiesChanged) updated += 1;
      if (pb.position !== nb.position || pb.parentId !== (nb.parentId ?? null)) moved += 1;
    }
    for (const pb of previous) {
      if (!nextById.has(pb.id)) deleted += 1;
    }
    return { inserted, updated, moved, deleted };
  }
}

/** The flush's `pages.version` CAS guard failed inside the transaction. */
class FlushCasConflict extends Error {
  constructor() {
    super("pages.version CAS conflict (concurrent external writer)");
  }
}