import type { SseBroker } from "../api/sse-broker";
import {
  collectSearchableSubjectFields,
  configuredPiiSubjectKms,
  decryptPiiFieldValues,
  isPiiCiphertext,
  PII_ERASED_SENTINEL,
} from "../crypto";
import type { DbRow } from "../db/connection";
import { tenantChannel } from "../engine/constants";
import type { EntityId, JobRunnerRef, Registry, SessionUser } from "../engine/types";
import type { SearchAdapter, SearchDocument } from "../search/types";
import type { EventConsumer } from "./event-dispatcher";

// --- Search Index Consumer (async, via event-dispatcher) ---
//
// Search-Indexierung läuft seit D.4 als async EventConsumer über den event-
// dispatcher, nicht mehr als synchroner postSave/postDelete-hook. Das
// spiegelt Marten's ISubscription-Pattern: ein einziger async Pfad für alle
// non-inline side-effects.
//
// Event → Search-Op Mapping:
//
//   <entity>.created     → index(tenantId, doc)
//   <entity>.updated     → index(tenantId, doc)   // re-index mit neuem state
//   <entity>.restored    → index(tenantId, doc)   // wiederbeleben
//   <entity>.deleted     → remove(tenantId, type, id)
//
// Der Document-State wird aus dem Event rekonstruiert (kein SaveContext
// mehr available). Regel:
//
//   created:  state = event.payload            // ganze entity ist im payload
//   updated:  state = { ...previous, ...changes }  // rekonstruiert neuen state
//   restored: state = event.payload.previous   // restored field-set
//
// Sensitive fields sind aus dem event log bereits gestrippt (event-store-
// executor.ts), also kriegt der Search-Index sie ebenfalls nicht — das ist
// die gleiche Garantie wie vorher beim postSave-hook.
//
// Batch-Variante gibt's aktuell nicht mehr — jeder Event triggert einen
// eigenen index()-call. Wenn Performance nach Scale-Messung das erfordert,
// kann der event-dispatcher später eine Batch-Handler-Variante bekommen.
export const SEARCH_CONSUMER_NAME = "system:consumer:search";

export function createSearchEventConsumer(
  searchAdapter: SearchAdapter,
  registry: Registry,
): EventConsumer {
  return {
    name: SEARCH_CONSUMER_NAME,
    // ponytail: count-based maxAttempts, not real backoff — retries hammer
    // the search adapter every pollIntervalMs during an outage instead of
    // spacing out. 1200 attempts * 100ms default poll ~= 2min budget for a
    // still-provisioning Meilisearch to come up at boot (was 10 = ~1s,
    // killing the consumer near-instantly). Upgrade path: time-based
    // dead-lettering in event-dispatcher-delivery.ts if 2min isn't enough.
    errorPolicy: { maxAttempts: 1200 },
    handler: async (event) => {
      const entityName = event.aggregateType;
      const verb = event.type.split(".").pop();
      const tenantId = event.tenantId;

      // skip: delete/forgotten remove the index entry — reconstruct only
      // makes sense for created/updated/restored (field data in payload).
      if (verb === "deleted" || verb === "forgotten") {
        await searchAdapter.remove(tenantId, entityName, event.aggregateId);
        return;
      }

      if (verb !== "created" && verb !== "updated" && verb !== "restored") {
        // skip: other event types (custom domain events, future verbs) don't
        // carry a search-indexable payload shape. If a future feature needs
        // them indexed, it registers its own multiStreamProjection.
        return;
      }

      let state = reconstructStateForSearch(event.payload, verb);
      state = await decryptSearchableSubjectFields(entityName, state, registry);
      // skip: erased subject — drop the doc so a rebuild cannot resurrect plaintext.
      if (hasErasedSearchableSubjectField(entityName, state, registry)) {
        await searchAdapter.remove(tenantId, entityName, event.aggregateId);
        return;
      }
      const doc = await buildSearchDocument(entityName, event.aggregateId, state, registry);
      if (!doc) {
        // skip: entity isn't searchable (no searchable fields declared)
        return;
      }
      await searchAdapter.index(tenantId, doc);
    },
  };
}

// #1610 — subject-annotated searchable fields are ciphertext in the event
// payload; decrypt into the derived index only. No KMS → omit ciphertext
// values rather than indexing blobs.
export async function decryptSearchableSubjectFields(
  entityName: string,
  state: Record<string, unknown>,
  registry: Registry,
): Promise<Record<string, unknown>> {
  const entity = registry.getEntity(entityName);
  if (!entity) return state;
  const fields = collectSearchableSubjectFields(entity);
  if (fields.length === 0) return state;
  const kms = configuredPiiSubjectKms();
  if (!kms) {
    const out = { ...state };
    for (const name of fields) {
      if (isPiiCiphertext(out[name])) delete out[name];
    }
    return out;
  }
  try {
    return await decryptPiiFieldValues(state, fields, kms, {
      requestId: "system:consumer:search",
    });
  } catch (err) {
    console.warn(
      `[kumiko:search] decryptSearchableSubjectFields failed for "${entityName}" — ` +
        `dropping ciphertext fields for this document instead of wedging the consumer.`,
      err,
    );
    const out = { ...state };
    for (const name of fields) {
      if (isPiiCiphertext(out[name])) delete out[name];
    }
    return out;
  }
}

export function hasErasedSearchableSubjectField(
  entityName: string,
  state: Record<string, unknown>,
  registry: Registry,
): boolean {
  const entity = registry.getEntity(entityName);
  if (!entity) return false;
  return collectSearchableSubjectFields(entity).some((name) => state[name] === PII_ERASED_SENTINEL);
}

// Rebuild the entity-state a search index needs from the event-payload alone.
// Three shapes to handle — see event-store-executor.ts for the emitter side.
function reconstructStateForSearch(
  payload: Record<string, unknown>,
  verb: "created" | "updated" | "restored",
): Record<string, unknown> {
  if (verb === "created") {
    // create: payload IS the entity (minus sensitive fields, already
    // stripped by event-store-executor)
    return payload;
  }
  if (verb === "updated") {
    // update: payload = { changes, previous }. Merge to get the new state
    // the index should reflect. Sensitive fields already filtered out.
    const previous = (payload["previous"] as Record<string, unknown> | undefined) ?? {}; // @cast-boundary engine-payload
    const changes = (payload["changes"] as Record<string, unknown> | undefined) ?? {}; // @cast-boundary engine-payload
    return { ...previous, ...changes };
  }
  // restored: payload = { previous }. The restored entity is whatever the
  // field-values were at delete time — restore copies them back verbatim.
  return (payload["previous"] as Record<string, unknown> | undefined) ?? {}; // @cast-boundary engine-payload
}

// buildSearchDocument runs per save — without dedup a colliding contributor
// key would spam one warn line per write on the hotpath. Scoped per registry
// (not module-global) so each app/test instance dedups independently: a
// process-wide Set would silence the warning for every later registry once any
// registry hit a given collision, and would leak dedup-state across tests.
const warnedKeyCollisionsByRegistry = new WeakMap<Registry, Set<string>>();
function warnOncePerKeyCollision(
  registry: Registry,
  entityName: string,
  key: string,
  isBaseField: boolean,
): void {
  let warned = warnedKeyCollisionsByRegistry.get(registry);
  if (!warned) {
    warned = new Set<string>();
    warnedKeyCollisionsByRegistry.set(registry, warned);
  }
  const dedupKey = `${entityName}:${key}`;
  // skip: already warned for this entity:key collision — dedup the hotpath
  if (warned.has(dedupKey)) return;
  warned.add(dedupKey);
  const collidesWith = isBaseField ? `base field "${key}"` : `earlier contributor key "${key}"`;
  console.warn(
    `[kumiko:search] searchPayloadExtension on "${entityName}" tried to overwrite ` +
      `${collidesWith} — keeping the first value. Rename the contributor key.`,
  );
}

// Build a SearchDocument from raw field-state. Parallel to the old
// buildSearchDocument that took a SaveContext — same selector logic, just
// a different input shape.
export async function buildSearchDocument(
  entityName: string,
  entityId: EntityId,
  state: Record<string, unknown>,
  registry: Registry,
): Promise<SearchDocument | null> {
  const entity = registry.getEntity(entityName);
  if (!entity) return null;

  const searchableFields = registry.getSearchableFields(entityName);
  const extensions = registry.getSearchPayloadExtensions(entityName);
  // Skip-Guard: kein indexable payload UND keine Extensions → kein doc.
  // Extensions können auch ohne searchable-Stammfields den index befüllen
  // (z.B. customFields-only-indexierung), daher muss die check beide
  // berücksichtigen.
  if (searchableFields.length === 0 && extensions.length === 0) return null;

  const embeddedFields = new Set<string>();
  for (const [fname, fdef] of Object.entries(entity.fields)) {
    if (fdef.type === "embedded") embeddedFields.add(fname);
  }

  const fields: Record<string, unknown> = {};
  for (const f of searchableFields) {
    const underscoreIdx = f.indexOf("_");
    if (underscoreIdx > 0) {
      const parentKey = f.slice(0, underscoreIdx);
      if (embeddedFields.has(parentKey)) {
        const subKey = f.slice(underscoreIdx + 1);
        const parent = state[parentKey];
        // A list-embedded parent contributes one indexed value per row —
        // Meilisearch indexes a string array as a searchable multi-value.
        if (Array.isArray(parent)) {
          const values = parent
            .filter((row): row is DbRow => Boolean(row) && typeof row === "object")
            .map((row) => row[subKey])
            .filter((value) => value !== undefined);
          if (values.length > 0) fields[f] = values;
        } else if (parent && typeof parent === "object") {
          const value = (parent as DbRow)[subKey];
          if (value !== undefined) fields[f] = value;
        }
        continue;
      }
    }
    if (state[f] !== undefined) {
      fields[f] = state[f];
    }
  }

  // F3 — Search-Payload-Extensions: contributors merge flat fields into the
  // search-doc (customFields-bundle / tags / computed-counts / etc.).
  // Sequential await — extensions are expected sync or sub-millisecond async;
  // sequential keeps the path simple and deterministic.
  //
  // Precedence is base-fields-win: a contributor key that collides with a
  // searchable Stammfield is dropped (not silently merged over the real value)
  // and warned. A jsonb custom-field that happens to share a Stammfield name
  // must not shadow the indexed Stammfield.
  const baseFieldKeys = new Set(Object.keys(fields));
  for (const contribute of extensions) {
    const contributed = await contribute({ entityName, entityId, state });
    for (const [key, value] of Object.entries(contributed)) {
      if (Object.hasOwn(fields, key)) {
        warnOncePerKeyCollision(registry, entityName, key, baseFieldKeys.has(key));
        continue;
      }
      fields[key] = value;
    }
  }

  return {
    entityType: entityName,
    entityId,
    weight: entity.searchWeight ?? 1,
    fields,
  };
}

// --- SSE Broadcast (async, via event-dispatcher) ---
//
// SSE-Broadcast läuft seit D.3 als async EventConsumer über den event-
// dispatcher, nicht mehr als synchroner postSave/postDelete-hook. Das hat
// zwei Konsequenzen:
//
// 1. **Event-native Payload-Shape.** Der SSE-event spiegelt den StoredEvent:
//    `type` ist event.type ("user.created", "unit.updated"), `data` enthält
//    id, aggregateType, version und die event-payload — keine künstliche
//    "system:event:<entity>:<verb>" Hülle mehr. Clients haben direkten
//    Zugriff auf `payload.changes` + `payload.previous` (wie im event-log).
// 2. **Eventual consistency statt Read-after-Write.** Ein SSE-Event kommt
//    ~10–100ms nach dem HTTP-200 (abhängig von pollIntervalMs). UI-Clients
//    die auf optimistic-update setzen merken das nicht; strictly-waiting
//    Clients müssten poll-after-write.
//
// Tests drain deterministisch via `await stack.eventDispatcher.runOnce()`.
export const SSE_BROADCAST_CONSUMER_NAME = "system:consumer:sse-broadcast";

export function createSseBroadcastEventConsumer(sseBroker: SseBroker): EventConsumer {
  return {
    name: SSE_BROADCAST_CONSUMER_NAME,
    // Per-instance delivery: each API process has its own pool of SSE
    // clients and its own cursor. Every instance reads every event
    // independently and pushes to its local clients. Without this, in a
    // split-deploy (API-1/API-2 + Worker, Welle 2.5), only ONE API
    // instance would pick up each event (shared-cursor SKIP LOCKED) and
    // the other instance's clients would never see updates. SSE clients
    // pin to a specific API process via the HTTP long-poll; cross-process
    // delivery isn't solvable by sharing a cursor.
    delivery: "per-instance",
    handler: async (event) => {
      sseBroker.pushToChannel(tenantChannel(event.tenantId), {
        type: event.type,
        data: {
          id: event.aggregateId,
          aggregateType: event.aggregateType,
          version: event.version,
          payload: event.payload,
          createdAt: event.createdAt,
        },
      });
    },
  };
}

// --- Job-Trigger Consumer (async, via event-dispatcher) ---
//
// r.job's `trigger.on` historically only fired via the synchronous
// write-handler dispatch path (dispatch-write.ts's afterCommitHooks calling
// jobRunner.handleEvent). Events appended any other way — an
// r.multiStreamProjection's ctx.unsafeAppendEvent, or a raw
// event-store-executor write (e.g. `files`' fileRef.created) — never
// reached it, so a job could never trigger on an r.defineEvent-registered
// event (kumiko-framework#1505).
//
// This consumer closes that gap the same way search/SSE do: read every
// committed event off the shared cursor and re-check job triggers. It is
// scoped EXACTLY to the gap — the write/query-handler-QN skip below is
// defense-in-depth: no stored event's `type` is a handler QN today (entity
// events are "entity.verb"; ctx.appendEvent enforces defineEvent-only
// ownership), but if that ever changes, this consumer must not re-fire a
// trigger the synchronous dispatch-write.ts path already handled.
//
// Delivery is at-least-once (cursor semantics) where the synchronous path
// is effectively once — job handlers reached via an r.defineEvent trigger
// must be idempotent (same expectation r.multiStreamProjection applies
// already carry).
export const JOB_TRIGGER_CONSUMER_NAME = "system:consumer:job-trigger";

export function createJobTriggerEventConsumer(
  jobRunner: JobRunnerRef,
  registry: Registry,
): EventConsumer {
  return {
    name: JOB_TRIGGER_CONSUMER_NAME,
    handler: async (event) => {
      // skip: write/query-handler QN — already dispatched synchronously by
      // dispatch-write.ts's afterCommitHooks. Re-firing here would
      // double-enqueue every existing handler-triggered job.
      if (registry.getWriteHandler(event.type) || registry.getQueryHandler(event.type)) return;
      // skip: no r.defineEvent registered under this type — nothing this
      // consumer is responsible for.
      if (!registry.getEvent(event.type)) return;

      const user: SessionUser = {
        id: event.metadata.userId,
        tenantId: event.tenantId,
        roles: [],
      };
      await jobRunner.handleEvent(event.type, event.payload, user);
    },
  };
}

// --- Access-Invalidation Consumer (async, via event-dispatcher) ---
//
// #1524 (Design), #1558 (channel + stream subscribe), #1559 (session-revoke
// emits an event). This consumer is the last leg: it watches for the two
// event types that can make an already-issued JWT stale mid-stream and
// pushes an invalidation through sseBroker.publishAccessInvalidation, which
// dispatch-stream.ts's subscribeAccessInvalidation listener turns into an
// AccessDeniedError thrown into the open stream.
//
// Event types are literal strings, not imports — this package (framework)
// cannot depend on bundled-features (sessions, tenant own these events),
// mirrors the es-ops-seed precedent (literal QNs, no subpath import).
//
//   "sessions:event:session-revoked" — payload.userId direct (own aggregate,
//     see bundled-features/sessions/session-revoked-event.ts). Fired for
//     both self-service revoke and the privileged cross-tenant
//     revoke-all-for-user (SYSTEM_TENANT_ID-anchored DSGVO Art.18 freeze).
//     fetchPendingEvents has no tenant predicate, so the SYSTEM_TENANT_ID-
//     anchored event reaches this consumer the same as any other — routing
//     is purely on payload.userId, never on event.tenantId.
//   "tenant-membership.updated" / "tenant-membership.deleted" — role change
//     or member removal. userId isn't in payload.changes (update only
//     carries the changed fields, e.g. { roles }) so it's read from
//     payload.previous, the full pre-write entity snapshot written by
//     createEventStoreExecutor. tenantMembershipEntity declares no
//     encrypted/PII fields, so previous.userId is plaintext — no KMS
//     decrypt step (see kumiko-framework#1560 PR discussion; the issue's
//     original handoff assumed encryption that doesn't apply here).
//
// Over-invalidation (e.g. tenant-membership.updated fired by something
// other than a role change) is fail-safe: the stream closes, the client
// reconnects and re-authorizes. No opt-out/allowlist — #1524 chose
// global-by-default specifically so this can't be silently disabled per
// handler.
//
// Scope note: entityEventName also emits "tenant-membership.forgotten" (DSGVO
// erasure) and ".restored" — .restored re-grants access so ignoring it is
// correct, .forgotten is not currently produced by any user-data-rights
// pipeline for memberships but would be revocation-relevant if it ever is.
// Out of #1560's stated scope (session-revoke + role/membership change),
// tracked rather than handled speculatively here.
export const ACCESS_INVALIDATION_CONSUMER_NAME = "system:consumer:access-invalidation";

const SESSION_REVOKED_EVENT_TYPE = "sessions:event:session-revoked";
const TENANT_MEMBERSHIP_UPDATED_EVENT_TYPE = "tenant-membership.updated";
const TENANT_MEMBERSHIP_DELETED_EVENT_TYPE = "tenant-membership.deleted";

function readUserIdFromPreviousSnapshot(payload: Record<string, unknown>): string | undefined {
  const previous = payload["previous"];
  if (typeof previous !== "object" || previous === null) return undefined;
  const userId = (previous as Record<string, unknown>)["userId"];
  return typeof userId === "string" && userId.length > 0 ? userId : undefined;
}

export function createAccessInvalidationEventConsumer(sseBroker: SseBroker): EventConsumer {
  return {
    name: ACCESS_INVALIDATION_CONSUMER_NAME,
    // Per-instance, same reasoning as SSE broadcast: subscribeAccessInvalidation
    // listeners live in this process's in-memory sseBroker only. A shared
    // cursor would deliver the event to exactly one instance and leave
    // every other instance's open streams for that user un-invalidated.
    delivery: "per-instance",
    handler: async (event) => {
      if (event.type === SESSION_REVOKED_EVENT_TYPE) {
        const userId = event.payload["userId"];
        // skip: malformed session-revoked payload — fail open on this one
        // event rather than dead-lettering the whole consumer (halt-on-
        // poison would otherwise permanently stop access-invalidation for
        // every user behind one bad row).
        if (typeof userId !== "string" || userId.length === 0) return;
        sseBroker.publishAccessInvalidation(userId);
      }

      if (
        event.type === TENANT_MEMBERSHIP_UPDATED_EVENT_TYPE ||
        event.type === TENANT_MEMBERSHIP_DELETED_EVENT_TYPE
      ) {
        const userId = readUserIdFromPreviousSnapshot(event.payload);
        // skip: previous snapshot missing/malformed userId — same fail-open
        // reasoning as above.
        if (userId === undefined) return;
        sseBroker.publishAccessInvalidation(userId);
      }
    },
  };
}