import { createHash } from "node:crypto";
import { Inject, Injectable, Optional, type OnModuleInit } from "@nestjs/common";
import type Redis from "ioredis";
import { z } from "zod";
import { EVENTS } from "../constants.js";
import { loadEnv } from "../config/env.js";
import { logger } from "./logger.js";
import { EventBus } from "./event-bus.js";
import { RedisService } from "./redis.service.js";

/**
 * TASK-0135 performance caching — the four hot-path caches with their exact
 * invalidation semantics (docs/development/perf-report.md):
 *
 *  - **page tree** — version-key: key `tree:{workspaceId}:{parentId}:{version}`
 *    where `version` is the workspace's content-version counter (a monotonic
 *    Redis counter bumped on EVERY page/block write in the workspace). Reads
 *    key by the current version; writes bump the version → a read can never
 *    observe a stale entry (the version in the key IS the freshness proof).
 *  - **permission decisions** — TTL 60 s (`PERM_CACHE_TTL_MS`) + event bust
 *    (`permission.changed` busts the granter/grantee's cache prefix + the
 *    resource-ancestor prefixes of the named resource). The TTL is the safety
 *    net: no event → the decision is at most 60 s stale.
 *  - **search results** — short TTL 5 s keyed by
 *    `{workspaceId, query-hash, filters-hash, principalId}` — permission
 *    relevant, so the principal is part of the key (two principals never share
 *    an entry). Only filtered results are cached.
 *  - **database column metadata** — TTL 60 s + bust on `database.updated`
 *    (property/view changes carry `propertyId`/`viewId` or `fields`).
 *
 * Every cached entry is a STRUCTURED value, re-validated with zod on read: a
 * corrupt/shape-mismatched entry is treated as a cache MISS (never a 500).
 *
 * The cache is OPTIONAL: Redis blips degrade to pass-through (the caller's
 * uncached path runs) — the API must not fail a request because the cache
 * layer is down (same posture as RedisService).
 */

export const CACHE_TTLS = {
  /** Permission decision TTL — the "no event → stale ≤ 60 s" bound. */
  PERMISSION_MS: 60_000,
  /** Database column-metadata TTL — busted on `database.updated`. */
  METADATA_MS: 60_000,
  /** Search-result TTL — short by design (freshness + permission keying). */
  SEARCH_MS: 5_000
} as const;

const PREFIX = {
  tree: "perf:tree",
  version: "perf:version",
  perm: "perf:perm",
  search: "perf:search",
  metadata: "perf:metadata"
} as const;

/** A structured, zod-validated cache entry. `schema` defines the shape. */
export interface StructuredEntry<T> {
  schema: z.ZodType<T>;
}

function key(prefix: string, ...parts: (string | number)[]): string {
  return [prefix, ...parts].join(":");
}

/** Stable short hash for the search key (query + filters). */
function hash(parts: unknown[]): string {
  return createHash("sha256").update(JSON.stringify(parts)).digest("hex").slice(0, 32);
}

@Injectable()
export class CacheService implements OnModuleInit {
  constructor(
    @Optional() @Inject(RedisService) private readonly redisProvider?: RedisService,
    @Optional() @Inject(EventBus) private readonly eventsProvider?: EventBus
  ) {}

  /**
   * Enablement from `PERF_CACHE` (`auto|on|off`): "auto" (default) = ON in
   * production only. Dev/test run UNCACHED by default — the event-invalidation
   * contract assumes every mutation flows through the event bus, which tests
   * (direct-Prisma writes) and dev tooling may bypass; the cache's own
   * integration suite forces `PERF_CACHE=on` to exercise the live semantics.
   */
  private readonly mode: "on" | "off" = (() => {
    const env = loadEnv();
    const on = env.PERF_CACHE === "on" || (env.PERF_CACHE === "auto" && env.NODE_ENV === "production");
    return on ? "on" : "off";
  })();

  /** Whether the cache layer is active for this process. */
  isOn(): boolean {
    return this.mode === "on";
  }

  onModuleInit(): void {
    if (this.isOn() && this.eventsProvider) this.wireInvalidation(this.eventsProvider);
  }

  private get redis(): Redis | null {
    try {
      return this.redisProvider?.getRedis() ?? null;
    } catch {
      return null;
    }
  }

  /**
   * Subscribe to the event bus for invalidation. Called once at module init
   * (the global `CacheService` instance) — see `wireInvalidation()`.
   */
  wireInvalidation(events: EventBus): void {
    // ---- permission bust: grant change → fresh on the NEXT decision.
    // `permission.changed` payloads (all 5 publish sites) carry
    // `{ workspaceId, userId, ... }` — the user whose grants changed
    // (workspace role, team add/remove, ownership transfer). `appliesToResource`
    // is a DIRECT id match (no ancestor walking), so a grant change only
    // alters decisions for THAT user in THAT workspace — bust the subject
    // prefix `perm:{ws}:{user}:*`. (A workspace-wide bust is the fallback if a
    // future payload omits `userId`; it stays bounded to one workspace.)
    // The 60 s TTL is the safety net for any event that is missed.
    events.onLocal(EVENTS.PERMISSION_CHANGED, (body) => {
      let payload: Record<string, unknown> = {};
      try {
        payload = JSON.parse(body.payload);
      } catch {
        return;
      }
      const workspaceId = payload.workspaceId as string | undefined;
      const userId = payload.userId as string | undefined;
      const r = this.redis;
      if (!r || !workspaceId) return;
      const patterns = [key(PREFIX.perm, workspaceId, userId ?? "*", "*")];
      this.scanDel(r, patterns);
    });

    // ---- content version bump: EVERY page/block write in a workspace.
    // `page.updated` (block + page writes, incl. the realtime flush) and
    // page create/delete bump the workspace's content version. (Database
    // record/property writes do NOT touch the page block tree, so
    // `database.updated` only busts the metadata cache — see below.)
    for (const event of [EVENTS.PAGE_UPDATED, EVENTS.PAGE_CREATED, EVENTS.PAGE_DELETED]) {
      events.onLocal(event, (body) => {
        let payload: Record<string, unknown> = {};
        try {
          payload = JSON.parse(body.payload);
        } catch {
          return;
        }
        const workspaceId = payload.workspaceId as string | undefined;
        if (workspaceId) void this.bumpVersion(workspaceId);
      });
    }

    // ---- metadata bust: property/view changes. `database.updated` carries
    // `databaseId` (always) + `propertyId`/`viewId`/`fields` (the specific
    // change). Bust the whole metadata namespace for the database (both the
    // property-list and view-settings entries) — a record or property change
    // can affect either view of the schema.
    events.onLocal(EVENTS.DATABASE_UPDATED, (body) => {
      let payload: Record<string, unknown> = {};
      try {
        payload = JSON.parse(body.payload);
      } catch {
        return;
      }
      const databaseId = payload.databaseId as string | undefined;
      const r = this.redis;
      if (databaseId && r) this.scanDel(r, [key(PREFIX.metadata, databaseId, "*")]);
    });
  }

  // ---------------------------------------------------------------------
  // content version (the tree cache's freshness token)
  // ---------------------------------------------------------------------

  /** Current workspace content version (0 when Redis is unavailable or off). */
  async getVersion(workspaceId: string): Promise<number> {
    const r = this.redis;
    if (!r || !this.isOn()) return 0;
    try {
      const v = await r.get(key(PREFIX.version, workspaceId));
      return v ? Number(v) : 0;
    } catch (err) {
      logger.warn(`cache.getVersion failed: ${(err as Error).message}`);
      return 0;
    }
  }

  /** Monotonic bump; returns the NEW version. */
  async bumpVersion(workspaceId: string): Promise<number> {
    const r = this.redis;
    if (!r) return 0;
    try {
      return await r.incr(key(PREFIX.version, workspaceId));
    } catch (err) {
      logger.warn(`cache.bumpVersion failed: ${(err as Error).message}`);
      return 0;
    }
  }

  // ---------------------------------------------------------------------
  // generic structured get/set (zod on read: corrupt entry = miss)
  // ---------------------------------------------------------------------

  async getStructured<T>(fullKey: string, entry: StructuredEntry<T>): Promise<T | null> {
    const r = this.redis;
    if (!r || !this.isOn()) return null;
    try {
      const raw = await r.get(fullKey);
      if (!raw) return null;
      let parsed: unknown;
      try {
        parsed = JSON.parse(raw);
      } catch {
        // corrupt JSON → miss (never a 500)
        await r.del(fullKey).catch(() => undefined);
        return null;
      }
      const result = entry.schema.safeParse(parsed);
      if (!result.success) {
        // shape mismatch → corrupt → miss
        await r.del(fullKey).catch(() => undefined);
        return null;
      }
      return result.data;
    } catch (err) {
      logger.warn(`cache.getStructured failed: ${(err as Error).message}`);
      return null;
    }
  }

  async setStructured(fullKey: string, value: unknown, ttlMs: number): Promise<void> {
    const r = this.redis;
    if (!r || !this.isOn()) return;
    try {
      await r.set(fullKey, JSON.stringify(value), "PX", ttlMs);
    } catch (err) {
      logger.warn(`cache.setStructured failed: ${(err as Error).message}`);
    }
  }

  private scanDel(r: Redis, patterns: string[]): void {
    for (const pattern of patterns) {
      void (async () => {
        let cursor = "0";
        do {
          const [next, keys] = await r.scan(cursor, "MATCH", pattern, "COUNT", 200).catch(() => ["0", []] as [string, string[]]);
          if (keys.length) await r.del(...keys).catch(() => undefined);
          cursor = next;
        } while (cursor !== "0");
      })().catch(() => undefined);
    }
  }

  // ---------------------------------------------------------------------
  // 1. page-tree cache (version-key)
  // ---------------------------------------------------------------------

  /**
   * Read the block tree for `pageId`. Freshness is guaranteed by the
   * version-key: the key embeds the workspace's content version at read
   * time, and every page/block write bumps that version — so a write that
   * lands before the read is invisible (new key), and a write that lands
   * after means the NEXT read uses a new key. No stale reads across
   * concurrent writers (the integration race test asserts this).
   */
  async getTree(
    workspaceId: string,
    pageId: string,
    loader: () => Promise<unknown[]>,
    schema: z.ZodType<unknown[]>
  ): Promise<{ data: unknown[]; cached: boolean }> {
    if (!this.isOn()) return { data: await loader(), cached: false };
    const version = await this.getVersion(workspaceId);
    if (version > 0) {
      const fullKey = key(PREFIX.tree, workspaceId, pageId, version);
      const hit = await this.getStructured(fullKey, { schema });
      if (hit) return { data: hit, cached: true };
      const data = await loader();
      await this.setStructured(fullKey, data, CACHE_TTLS.PERMISSION_MS);
      return { data, cached: false };
    }
    return { data: await loader(), cached: false };
  }

  // ---------------------------------------------------------------------
  // 2. permission-decision cache (TTL 60 s + event bust)
  // ---------------------------------------------------------------------

  /**
   * Cache a permission decision. Key = (workspace, user, scope) where scope
   * hashes the named resource ids (most specific → the decision for that
   * resource; absent → the workspace baseline). Bust on
   * `permission.changed` (subject + resource prefixes), TTL 60 s otherwise.
   */
  async getPermission(
    workspaceId: string,
    userId: string,
    scope: { pageId?: string; databaseId?: string; recordId?: string; fileId?: string; teamId?: string } | undefined,
    loader: () => Promise<string>,
    schema: z.ZodType<string>
  ): Promise<{ level: string; cached: boolean }> {
    const scopeHash = hash(Object.entries(scope ?? {}).filter(([, v]) => v != null).map(([k, v]) => [k, v]));
    const fullKey = key(PREFIX.perm, workspaceId, userId, scopeHash);
    const hit = await this.getStructured(fullKey, { schema });
    if (hit) return { level: hit, cached: true };
    const level = await loader();
    await this.setStructured(fullKey, level, CACHE_TTLS.PERMISSION_MS);
    return { level, cached: false };
  }

  // ---------------------------------------------------------------------
  // 3. search-result cache (TTL 5 s, permission-keyed by principal)
  // ---------------------------------------------------------------------

  async getSearch(
    workspaceId: string | undefined,
    principalId: string,
    input: Record<string, unknown>,
    loader: () => Promise<unknown>,
    schema: z.ZodType<unknown>
  ): Promise<{ data: unknown; cached: boolean }> {
    // Only permission-relevant (filtered) results: key by the full input
    // (query + types + filters + cursor) AND the principal.
    const { q, ...filters } = input;
    const fullKey = key(
      PREFIX.search,
      workspaceId ?? "global",
      principalId,
      hash([String(q ?? ""), filters])
    );
    const hit = await this.getStructured(fullKey, { schema });
    if (hit) return { data: hit, cached: true };
    const data = await loader();
    await this.setStructured(fullKey, data, CACHE_TTLS.SEARCH_MS);
    return { data, cached: false };
  }

  // ---------------------------------------------------------------------
  // 4. database column-metadata cache (TTL 60 s + bust on db.updated)
  // ---------------------------------------------------------------------

  async getMetadata(
    databaseId: string,
    kind: "properties" | "views",
    loader: () => Promise<unknown>,
    schema: z.ZodType<unknown>
  ): Promise<{ data: unknown; cached: boolean }> {
    const fullKey = key(PREFIX.metadata, databaseId, kind);
    const hit = await this.getStructured(fullKey, { schema });
    if (hit) return { data: hit, cached: true };
    const data = await loader();
    await this.setStructured(fullKey, data, CACHE_TTLS.METADATA_MS);
    return { data, cached: false };
  }
}