import {
  constraintOf,
  isLockNotAvailable,
  isTableAlreadyExists,
  isUniqueViolation,
} from "../pg-error";
import type { AnyDb } from "../query";
import { asRawClient, unsafeReadRetrying } from "../query";

/** NOTIFY on commit — wakes LISTEN subscribers (event-dispatcher). */
export async function notifyPgChannel(db: AnyDb, channel: string): Promise<void> {
  await asRawClient(db).unsafe(`SELECT pg_notify($1, '')`, [channel]);
}

// Tenant-scoped partial unique index over metadata.idempotencyKey.
// Expression index straight on the jsonb column — no dedicated key column,
// so it needs no INSERT-path change and covers admin-api's raw appends too
// (same metadata jsonb). CREATE ... IF NOT EXISTS makes this safe to call
// on every boot, same "ensure" pattern as ensureSnapshotVersionColumn: heals
// installs that predate the index without a table rebuild.
//
// CONCURRENTLY (not a plain CREATE): a non-concurrent build takes a SHARE
// lock for the full table scan on kumiko_events — the hottest table in the
// framework — blocking every append() for however long that scan takes on
// an existing installation's event history. CONCURRENTLY avoids that at the
// cost of needing to tolerate two failure modes a plain build doesn't have.
// Neither CREATE nor DROP ... CONCURRENTLY may run inside a transaction —
// no caller of this function (dev-server, schema-cli.ts, stack/db.ts) may
// wrap it in one, or Postgres raises 25001.
// undefined = index doesn't exist at all (nothing to drop, CREATE below
// handles it); false = exists but INVALID (crashed mid-build, needs DROP +
// rebuild); true = exists and valid.
async function indexValidity(client: ReturnType<typeof asRawClient>): Promise<boolean | undefined> {
  const rows = await client.unsafe(
    `SELECT i.indisvalid FROM pg_class c JOIN pg_index i ON i.indexrelid = c.oid ` +
      `WHERE c.relname = 'events_idempotency_uq' AND i.indrelid = '"kumiko_events"'::regclass`,
  );
  return (rows[0] as { indisvalid?: boolean } | undefined)?.indisvalid;
}

export async function ensureIdempotencyKeyIndex(db: AnyDb): Promise<void> {
  const client = asRawClient(db);

  try {
    // 1) A prior CONCURRENTLY build that got killed mid-flight (crash, deploy
    //    restart) leaves an INVALID index: the catalog entry exists, so
    //    IF NOT EXISTS below would silently skip forever, but the index is
    //    incomplete and not plannable for queries — it does NOT mean the
    //    constraint enforces nothing; Postgres keeps maintaining an INVALID
    //    index on every insert, it just refuses to use it for planning.
    //    Detect + rebuild it. Scoped to kumiko_events specifically (indrelid),
    //    not just the relname, so a same-named index in another schema can't
    //    false-positive this DROP.
    if ((await indexValidity(client)) === false) {
      await client.unsafe(`DROP INDEX CONCURRENTLY IF EXISTS "events_idempotency_uq"`);
    }

    await client.unsafe(
      `CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS "events_idempotency_uq" ON "kumiko_events" ` +
        `("tenant_id", (("metadata"->>'idempotencyKey'))) ` +
        `WHERE "metadata"->>'idempotencyKey' IS NOT NULL`,
    );
  } catch (e) {
    if (isBenignConcurrentIndexBuildRace(e)) {
      // "Benign" only means the losing side of a race, not that the index
      // actually landed valid — a lock_timeout on the CREATE (55P03) backs
      // off the same way a genuine duplicate-build race does, but leaves no
      // valid index at all. Re-check before declaring victory instead of
      // trusting the error class alone.
      // skip: sibling pod already built a valid index — this race loser is done.
      if ((await indexValidity(client)) === true) return;
      console.warn(
        `ensureIdempotencyKeyIndex: backed off on a "benign" race (${String(e)}) but ` +
          `"events_idempotency_uq" is still missing/invalid afterward — likely a ` +
          `lock_timeout during CREATE INDEX CONCURRENTLY, not an actual winner/loser race.`,
      );
      throw e;
    }
    throw duplicateIdempotencyKeyErrorOr(e);
  }
}

// Two pods booting concurrently against the same DB (rolling deploy): both
// see the index missing/invalid and both start a DROP/CREATE CONCURRENTLY
// build. The loser typically does NOT get the plain duplicate-relation
// no-op IF NOT EXISTS normally gives — it can instead see a unique-violation
// on Postgres' own pg_class catalog insert (23505, constraint
// pg_class_relname_nsp_index) or a lock-not-available (55P03) from the
// racing DDL. Both are benign: the other pod's build wins and this one just
// backs off. A 23505 against "events_idempotency_uq" itself is NOT this
// race — see duplicateIdempotencyKeyErrorOr.
function isBenignConcurrentIndexBuildRace(e: unknown): boolean {
  if (isTableAlreadyExists(e) || isLockNotAvailable(e)) return true;
  return isUniqueViolation(e) && constraintOf(e) === "pg_class_relname_nsp_index";
}

// A 23505 against "events_idempotency_uq" means real duplicate
// metadata->>'idempotencyKey' values for the same tenant already exist —
// CONCURRENTLY still enforces uniqueness on live inserts against the
// not-yet-valid index. That needs an operator to find + resolve the
// duplicates, not a crash-loop on every subsequent boot, so re-throw a
// distinguishable error instead of the raw driver error.
function duplicateIdempotencyKeyErrorOr(e: unknown): unknown {
  if (isUniqueViolation(e) && constraintOf(e) === "events_idempotency_uq") {
    return new Error(
      "ensureIdempotencyKeyIndex: duplicate metadata->>'idempotencyKey' values exist for at least " +
        "one tenant in kumiko_events — CREATE UNIQUE INDEX CONCURRENTLY cannot complete. Find and " +
        "resolve the duplicate idempotencyKey rows, then restart to retry.",
      { cause: e },
    );
  }
  return e;
}

export type SubsequentEventInsertParams = {
  readonly aggregateId: string;
  readonly aggregateType: string;
  readonly tenantId: string;
  readonly newVersion: number;
  readonly type: string;
  readonly eventVersion: number;
  // Plain objects, NOT pre-stringified JSON: Bun.SQL encodes a JS string
  // bound to ::jsonb as a JSON string scalar (double-encoding) — every
  // version>1 event written between 2026-05-25 and this fix carried
  // payload/metadata as jsonb strings instead of objects.
  readonly payload: Record<string, unknown>;
  readonly metadata: Record<string, unknown>;
  readonly createdBy: string;
  readonly expectedVersion: number;
};

export type SubsequentEventInsertRow = {
  readonly id: string | bigint;
  readonly created_at: Date | string;
};

/** INSERT … SELECT … WHERE EXISTS predecessor — stays raw (typed builder can't express this). */
export async function insertSubsequentEventRow(
  db: AnyDb,
  params: SubsequentEventInsertParams,
): Promise<SubsequentEventInsertRow | undefined> {
  const rows = (await asRawClient(db).unsafe(
    `INSERT INTO "kumiko_events" (
       aggregate_id, aggregate_type, tenant_id, version,
       type, event_version, payload, metadata, created_by
     )
     SELECT $1::uuid, $2, $3::uuid, $4,
            $5, $6, $7::jsonb,
            $8::jsonb, $9
     WHERE EXISTS (
       SELECT 1 FROM "kumiko_events"
       WHERE aggregate_id = $1::uuid
         AND version = $10
         AND tenant_id = $3::uuid
     )
     RETURNING id, created_at`,
    [
      params.aggregateId,
      params.aggregateType,
      params.tenantId,
      params.newVersion,
      params.type,
      params.eventVersion,
      params.payload,
      params.metadata,
      params.createdBy,
      params.expectedVersion,
    ],
  )) as ReadonlyArray<SubsequentEventInsertRow>;
  return rows[0];
}

export async function selectStreamMaxVersion(
  db: AnyDb,
  aggregateId: string,
  tenantId: string,
): Promise<number> {
  const rows = (await unsafeReadRetrying(
    db,
    `SELECT MAX("version") AS v FROM "kumiko_events" WHERE "aggregate_id" = $1 AND "tenant_id" = $2`,
    [aggregateId, tenantId],
  )) as ReadonlyArray<{ v: number | null }>;
  return rows[0]?.v ?? 0;
}

/** MAX(version) for one aggregate stream — no tenant filter (seed idempotency). */
export async function selectAggregateMaxVersion(db: AnyDb, aggregateId: string): Promise<number> {
  const rows = (await unsafeReadRetrying(
    db,
    `SELECT MAX("version") AS v FROM "kumiko_events" WHERE "aggregate_id" = $1`,
    [aggregateId],
  )) as ReadonlyArray<{ v: number | null }>;
  return rows[0]?.v ?? 0;
}

export async function selectEventsHighWaterMark(db: AnyDb): Promise<bigint> {
  const rows = (await unsafeReadRetrying(
    db,
    `SELECT COALESCE(MAX("id"), 0)::bigint AS max FROM "kumiko_events"`,
    [],
  )) as ReadonlyArray<{ max: bigint | string | number | null }>;
  const raw = rows[0]?.max;
  if (typeof raw === "bigint") return raw;
  if (raw === null || raw === undefined) return 0n;
  return BigInt(raw);
}

/** Head event id for lag metrics — alias for selectEventsHighWaterMark. */
// @wrapper-known semantic-alias
export async function selectEventsHeadId(db: AnyDb): Promise<bigint> {
  return selectEventsHighWaterMark(db);
}

export async function selectNextEventIdAfter(db: AnyDb, afterId: bigint): Promise<bigint | null> {
  const rows = (await unsafeReadRetrying(
    db,
    `SELECT "id" FROM "kumiko_events" WHERE "id" > $1 ORDER BY "id" ASC LIMIT 1`,
    [afterId],
  )) as ReadonlyArray<{ id: string | bigint }>;
  const row = rows[0];
  if (!row) return null;
  return typeof row.id === "bigint" ? row.id : BigInt(row.id);
}

export type SaveSnapshotParams = {
  readonly aggregateId: string;
  readonly tenantId: string;
  readonly aggregateType: string;
  readonly version: number;
  // Plain object — see SubsequentEventInsertParams on why pre-stringified
  // JSON double-encodes under Bun.SQL's ::jsonb binding.
  readonly state: Record<string, unknown>;
  readonly snapshotVersion: number;
};

// kumiko_snapshots predates snapshot_version — idempotent heal for existing
// installs, run from the same ensure path as table creation.
export async function ensureSnapshotVersionColumn(db: AnyDb): Promise<void> {
  await asRawClient(db).unsafe(
    `ALTER TABLE "kumiko_snapshots" ADD COLUMN IF NOT EXISTS "snapshot_version" integer NOT NULL DEFAULT 1`,
  );
}

export async function upsertSnapshot(db: AnyDb, params: SaveSnapshotParams): Promise<void> {
  await asRawClient(db).unsafe(
    `INSERT INTO "kumiko_snapshots"
       ("aggregate_id", "tenant_id", "aggregate_type", "version", "state", "snapshot_version")
     VALUES ($1, $2, $3, $4, $5::jsonb, $6)
     ON CONFLICT ("aggregate_id", "version") DO UPDATE SET
       "state" = $5::jsonb,
       "aggregate_type" = $3,
       "snapshot_version" = $6,
       "created_at" = now()`,
    [
      params.aggregateId,
      params.tenantId,
      params.aggregateType,
      params.version,
      params.state,
      params.snapshotVersion,
    ],
  );
}

export type ArchiveStreamParams = {
  readonly tenantId: string;
  readonly aggregateId: string;
  readonly aggregateType: string;
  readonly archivedBy: string;
  readonly reason: string | null;
};

export async function upsertArchivedStream(db: AnyDb, params: ArchiveStreamParams): Promise<void> {
  await asRawClient(db).unsafe(
    `INSERT INTO "kumiko_archived_streams"
       ("tenant_id", "aggregate_id", "aggregate_type", "archived_by", "reason")
     VALUES ($1, $2, $3, $4, $5)
     ON CONFLICT ("tenant_id", "aggregate_id") DO UPDATE SET
       "archived_at" = now(),
       "archived_by" = $4,
       "aggregate_type" = $3,
       "reason" = $5`,
    [params.tenantId, params.aggregateId, params.aggregateType, params.archivedBy, params.reason],
  );
}