import type {
  AuthorizationActor,
  AuthorizationScope,
  EkklesiaUserId,
  GrantId,
  GrantSourceKind,
  GrantSourceRef,
  PermissionId,
} from "@ekklesia/contracts";
import {
  isUserActor,
  parseAuthorizationScope,
  parseEkklesiaUserId,
  parseGrantId,
  parsePermissionId,
} from "@ekklesia/contracts";
import { and, desc, eq, gt, isNull, or, sql } from "drizzle-orm";
import type { AuthorizationDrizzleDatabase } from "./database.js";
import type {
  ActiveGrantQuery,
  AuthorizationGrantPort,
  AuthorizationPlatformOwnerPort,
} from "./ports.js";
import {
  ekklesiaCoreAuthorizationAudit,
  ekklesiaCoreGrant,
  ekklesiaCorePlatformOwner,
} from "./schema.js";

export type AuthorizationStoreErrorCode =
  | "AUTHORIZATION_GRANT_ALREADY_EXISTS"
  | "AUTHORIZATION_GRANT_CREATE_FAILED"
  | "AUTHORIZATION_GRANT_EXPIRATION_INVALID"
  | "AUTHORIZATION_GRANT_LIMIT_INVALID"
  | "AUTHORIZATION_GRANT_LIST_FAILED"
  | "AUTHORIZATION_GRANT_LOOKUP_FAILED"
  | "AUTHORIZATION_GRANT_NOT_FOUND"
  | "AUTHORIZATION_GRANT_REVOKE_FAILED"
  | "AUTHORIZATION_PLATFORM_OWNER_LOOKUP_FAILED"
  | "AUTHORIZATION_PLATFORM_OWNER_MUTATION_FAILED"
  | "AUTHORIZATION_PLATFORM_OWNER_NOT_FOUND";

const storeErrorMessages: Readonly<Record<AuthorizationStoreErrorCode, string>> = Object.freeze({
  AUTHORIZATION_GRANT_ALREADY_EXISTS:
    "An equivalent authorization grant is already effective for this user.",
  AUTHORIZATION_GRANT_CREATE_FAILED: "The authorization grant could not be created.",
  AUTHORIZATION_GRANT_EXPIRATION_INVALID:
    "The authorization grant expiration must be a valid instant in the future.",
  AUTHORIZATION_GRANT_LIMIT_INVALID: "The authorization grant page size is invalid.",
  AUTHORIZATION_GRANT_LIST_FAILED: "The authorization grant list could not be retrieved.",
  AUTHORIZATION_GRANT_LOOKUP_FAILED: "The authorization grant lookup could not be completed.",
  AUTHORIZATION_GRANT_NOT_FOUND: "The authorization grant was not found or is already revoked.",
  AUTHORIZATION_GRANT_REVOKE_FAILED: "The authorization grant could not be revoked.",
  AUTHORIZATION_PLATFORM_OWNER_LOOKUP_FAILED: "The platform-owner lookup could not be completed.",
  AUTHORIZATION_PLATFORM_OWNER_MUTATION_FAILED: "The platform-owner change could not be recorded.",
  AUTHORIZATION_PLATFORM_OWNER_NOT_FOUND:
    "The platform-owner record was not found or is already revoked.",
});

/** The largest page any caller may request from `listGrantsForUser`, enforced identically by the
 * store and by the HTTP surface so there is exactly one bound, not two that could drift. */
export const AUTHORIZATION_GRANT_LIST_MAX_LIMIT = 100;

/** Never carries the underlying database error, SQL text, host, port, path, or stack. */
export class AuthorizationStoreError extends Error {
  readonly code: AuthorizationStoreErrorCode;

  constructor(code: AuthorizationStoreErrorCode) {
    super(storeErrorMessages[code]);
    this.name = "AuthorizationStoreError";
    this.code = code;
    this.stack = `${this.name}: ${this.message}`;
    Object.freeze(this);
  }
}

function isAuthorizationStoreError(error: unknown): error is AuthorizationStoreError {
  return error instanceof AuthorizationStoreError;
}

export interface CreateGrantInput {
  readonly userId: EkklesiaUserId;
  readonly permission: PermissionId;
  readonly scope?: AuthorizationScope;
  readonly sourceKind: GrantSourceKind;
  readonly sourceRef?: GrantSourceRef;
  readonly actor: AuthorizationActor;
  /** Absent means the grant never expires. */
  readonly expiresAt?: Date;
}

export interface GrantRecord {
  readonly id: GrantId;
  readonly userId: EkklesiaUserId;
  readonly permission: PermissionId;
  readonly scope?: AuthorizationScope;
  readonly sourceKind: GrantSourceKind;
  readonly sourceRef?: GrantSourceRef;
  readonly createdAt: Date;
  readonly expiresAt?: Date;
  readonly revokedAt?: Date;
}

export interface RevokeGrantInput {
  readonly grantId: GrantId;
  readonly actor: AuthorizationActor;
  /** When provided, the revoke only matches a grant whose `sourceKind` equals this value — a
   * grant of a different source kind is treated exactly like a nonexistent one
   * (`AUTHORIZATION_GRANT_NOT_FOUND`). The generic explicit-grant HTTP API always passes
   * `"explicit"` here, so it can never revoke a relationship-derived grant regardless of caller
   * intent; internal Core callers that omit this may revoke any grant, matching pass 1's
   * original, unconstrained behavior. */
  readonly sourceKind?: GrantSourceKind;
}

/** One row's position in the `created_at DESC, id DESC` ordering, used as a strict keyset
 * boundary. Both values are required: comparing on the timestamp alone would skip or repeat rows
 * whenever two grants share a `created_at`, which concurrent creation makes routine. */
export interface GrantPageBoundary {
  readonly createdAt: Date;
  readonly id: GrantId;
}

export interface ListGrantsForUserOptions {
  /** Required. The maximum number of rows to return, 1 through
   * `AUTHORIZATION_GRANT_LIST_MAX_LIMIT`. There is deliberately no default and no unbounded
   * fallback anywhere: an unbounded listing returned a user's entire authorization history in one
   * response, and a bound a caller may simply omit is not a bound. */
  readonly limit: number;
  /** When provided, only grants strictly *after* this position in the ordering are returned. */
  readonly after?: GrantPageBoundary;
  /** When provided, only grants with this exact `sourceKind` are returned. */
  readonly sourceKind?: GrantSourceKind;
}

export interface GrantPage {
  readonly grants: readonly GrantRecord[];
  /** The last returned row's position when more rows exist, otherwise `undefined`. */
  readonly nextBoundary?: GrantPageBoundary;
}

export interface CreateAuthorizationStoreOptions {
  /** Injected only so expiration validation is deterministic under test. Production composition
   * leaves this unset and the store reads real time. */
  readonly now?: () => Date;
}

/**
 * Core's Postgres-backed authorization persistence: grants, platform-owner status, and the
 * append-only audit log, all behind one explicit factory — no environment reads, no connection
 * attempt, and no query merely from importing this module or calling this factory; every method
 * call is the only thing that ever touches PostgreSQL. Implements the narrow read ports
 * `createAuthorizationEvaluator` depends on directly, so the same store instance is both Core's
 * write API and the evaluator's read dependency.
 */
export interface AuthorizationStore extends AuthorizationGrantPort, AuthorizationPlatformOwnerPort {
  createGrant(input: CreateGrantInput): Promise<GrantRecord>;
  revokeGrant(input: RevokeGrantInput): Promise<void>;
  /** One bounded page of the user's grants, active or revoked, ordered `created_at DESC, id DESC`.
   * Never includes another user's grants. */
  listGrantsForUser(userId: EkklesiaUserId, options: ListGrantsForUserOptions): Promise<GrantPage>;
  grantPlatformOwner(userId: EkklesiaUserId, actor: AuthorizationActor): Promise<void>;
  revokePlatformOwner(userId: EkklesiaUserId, actor: AuthorizationActor): Promise<void>;
}

function actorId(actor: AuthorizationActor): string | null {
  return isUserActor(actor) ? actor.userId : null;
}

interface GrantRow {
  readonly id: string;
  readonly userId: string;
  readonly permissionId: string;
  readonly scope: string | null;
  readonly sourceKind: string;
  readonly sourceRef: string | null;
  readonly createdAt: Date;
  readonly expiresAt: Date | null;
  readonly revokedAt: Date | null;
}

/** Re-validates a row this same store just wrote, rather than trusting raw column values by
 * construction — defense in depth against a future schema/migration drift, matching the
 * paranoid-by-default posture the rest of Core's persistence code already follows. */
function toGrantRecord(row: GrantRow): GrantRecord {
  return Object.freeze({
    createdAt: row.createdAt,
    id: parseGrantId(row.id),
    permission: parsePermissionId(row.permissionId),
    ...(row.scope === null ? {} : { scope: parseAuthorizationScope(row.scope) }),
    ...(row.expiresAt === null ? {} : { expiresAt: row.expiresAt }),
    ...(row.revokedAt === null ? {} : { revokedAt: row.revokedAt }),
    sourceKind: row.sourceKind as GrantSourceKind,
    ...(row.sourceRef === null ? {} : { sourceRef: row.sourceRef }),
    userId: parseEkklesiaUserId(row.userId),
  });
}

/**
 * The five values that make two grant issuances the *same authorization source*: subject,
 * permission, scope, source kind, and source reference — with `NULL` equal only to `NULL`.
 * `expires_at`, creation metadata, actor metadata, and the grant id deliberately do **not**
 * distinguish them: re-issuing the identical authority with a different expiry or by a different
 * administrator is the same grant, not a second one.
 */
interface GrantEquivalence {
  readonly userId: string;
  readonly permissionId: string;
  readonly scope: string | null;
  readonly sourceKind: string;
  readonly sourceRef: string | null;
}

function equivalenceOf(input: CreateGrantInput): GrantEquivalence {
  return {
    permissionId: input.permission,
    scope: input.scope ?? null,
    sourceKind: input.sourceKind,
    sourceRef: input.sourceRef ?? null,
    userId: input.userId,
  };
}

/**
 * Serializes every create/revoke touching one equivalence tuple against a transaction-scoped
 * PostgreSQL advisory lock, so concurrent callers cannot both observe "no effective equivalent"
 * and both insert. A process-local mutex would not do this — nothing constrains the number of API
 * processes — and neither would an unlocked check-then-insert.
 *
 * The key is a 64-bit hash of the tuple encoded as a JSON array, which keeps `NULL` and the empty
 * string distinct and makes separator forgery through a crafted scope or source reference
 * impossible. A hash collision is harmless by construction: it can only make two unrelated tuples
 * serialize against each other, never make a real conflict go unnoticed, because the authoritative
 * decision is always the exact SQL predicate below and never the hash.
 *
 * `pg_advisory_xact_lock(bigint)` uses PostgreSQL's *single-key* advisory-lock space, which is
 * verified to be disjoint from the two-key space used by the migration runner
 * (`core/src/migrations/postgres-database.ts`) and by platform-owner bootstrap. Grant serialization
 * therefore can never block, or be blocked by, migration or bootstrap locking. Being
 * transaction-scoped, it is released on COMMIT or ROLLBACK with no manual unlock.
 */
async function lockEquivalence(
  tx: Parameters<Parameters<AuthorizationDrizzleDatabase["transaction"]>[0]>[0],
  equivalence: GrantEquivalence,
): Promise<void> {
  await tx.execute(
    sql`select pg_advisory_xact_lock(hashtextextended(jsonb_build_array(
      ${equivalence.userId}::text,
      ${equivalence.permissionId}::text,
      ${equivalence.scope}::text,
      ${equivalence.sourceKind}::text,
      ${equivalence.sourceRef}::text
    )::text, 0))`,
  );
}

/** Matches exactly the rows sharing an equivalence tuple. `is not distinct from` is what gives
 * `NULL = NULL` for the two nullable members; plain `=` would silently match nothing. */
function equivalenceCondition(equivalence: GrantEquivalence) {
  return and(
    eq(ekklesiaCoreGrant.userId, equivalence.userId),
    eq(ekklesiaCoreGrant.permissionId, equivalence.permissionId),
    sql`${ekklesiaCoreGrant.scope} is not distinct from ${equivalence.scope}::text`,
    eq(ekklesiaCoreGrant.sourceKind, equivalence.sourceKind),
    sql`${ekklesiaCoreGrant.sourceRef} is not distinct from ${equivalence.sourceRef}::text`,
  );
}

/** A supplied expiration must be a real instant strictly later than the acceptance clock. A past
 * or equal-to-now expiration produces a grant that is dead the moment it is written, which is
 * never what a caller meant; rejecting it here means trusted internal callers get the same
 * guarantee the HTTP surface enforces, rather than each validating separately. */
function assertValidExpiration(expiresAt: Date | undefined, now: Date): void {
  if (expiresAt === undefined) {
    return;
  }
  if (
    !(expiresAt instanceof Date) ||
    Number.isNaN(expiresAt.getTime()) ||
    expiresAt.getTime() <= now.getTime()
  ) {
    throw new AuthorizationStoreError("AUTHORIZATION_GRANT_EXPIRATION_INVALID");
  }
}

function assertValidLimit(limit: number): number {
  if (
    typeof limit !== "number" ||
    !Number.isSafeInteger(limit) ||
    limit < 1 ||
    limit > AUTHORIZATION_GRANT_LIST_MAX_LIMIT
  ) {
    throw new AuthorizationStoreError("AUTHORIZATION_GRANT_LIMIT_INVALID");
  }
  return limit;
}

export function createAuthorizationStore(
  db: AuthorizationDrizzleDatabase,
  options: CreateAuthorizationStoreOptions = {},
): AuthorizationStore {
  const now = options.now ?? ((): Date => new Date());

  return Object.freeze({
    createGrant: async (input: CreateGrantInput): Promise<GrantRecord> => {
      // Validated before any transaction opens, so a rejected expiration performs no database work
      // at all and can never leave a grant or an audit row behind.
      assertValidExpiration(input.expiresAt, now());
      const equivalence = equivalenceOf(input);
      try {
        return await db.transaction(async (tx) => {
          await lockEquivalence(tx, equivalence);

          // Under the lock, "currently effective" is authoritative: an equivalent that is revoked,
          // or whose expiry has already passed, does not block a fresh issuance.
          const effective = await tx
            .select({ id: ekklesiaCoreGrant.id })
            .from(ekklesiaCoreGrant)
            .where(
              and(
                equivalenceCondition(equivalence),
                isNull(ekklesiaCoreGrant.revokedAt),
                or(
                  isNull(ekklesiaCoreGrant.expiresAt),
                  gt(ekklesiaCoreGrant.expiresAt, sql`now()`),
                ),
              ),
            )
            .limit(1);
          if (effective.length > 0) {
            throw new AuthorizationStoreError("AUTHORIZATION_GRANT_ALREADY_EXISTS");
          }

          const [row] = await tx
            .insert(ekklesiaCoreGrant)
            .values({
              createdByActorId: actorId(input.actor),
              createdByActorType: input.actor.type,
              expiresAt: input.expiresAt ?? null,
              permissionId: input.permission,
              scope: input.scope ?? null,
              sourceKind: input.sourceKind,
              sourceRef: input.sourceRef ?? null,
              userId: input.userId,
            })
            .returning();
          if (!row) {
            throw new Error("grant insert returned no row");
          }

          await tx.insert(ekklesiaCoreAuthorizationAudit).values({
            action: "grant.create",
            actorId: actorId(input.actor),
            actorType: input.actor.type,
            grantId: row.id,
            permissionId: input.permission,
            scope: input.scope ?? null,
            subjectUserId: input.userId,
          });

          return toGrantRecord(row);
        });
      } catch (error) {
        if (isAuthorizationStoreError(error)) {
          throw error;
        }
        throw new AuthorizationStoreError("AUTHORIZATION_GRANT_CREATE_FAILED");
      }
    },

    /**
     * Revoking one grant revokes *every* currently unrevoked issuance sharing its equivalence
     * tuple, in one transaction, with one `grant.revoke` audit row per grant actually changed.
     *
     * Addressing a single id and revoking only that row was actively misleading: an operator who
     * revoked a grant received `success` while an equivalent duplicate kept authorizing the same
     * user. Expired-but-unrevoked equivalents are included deliberately — they are already
     * ineffective, but leaving them unrevoked would keep the history in two states for what is one
     * authorization source. Grants with a different subject, permission, scope, source kind, or
     * source reference are untouched.
     */
    revokeGrant: async (input: RevokeGrantInput): Promise<void> => {
      try {
        await db.transaction(async (tx) => {
          // This first read exists only to discover the equivalence tuple, which is safe to read
          // before the lock because those columns are immutable once written. It deliberately does
          // *not* settle whether the revocation may proceed: everything it observed can change
          // while this transaction waits for the lock below.
          const [target] = await tx
            .select({
              permissionId: ekklesiaCoreGrant.permissionId,
              scope: ekklesiaCoreGrant.scope,
              sourceKind: ekklesiaCoreGrant.sourceKind,
              sourceRef: ekklesiaCoreGrant.sourceRef,
              userId: ekklesiaCoreGrant.userId,
            })
            .from(ekklesiaCoreGrant)
            .where(
              and(
                eq(ekklesiaCoreGrant.id, input.grantId),
                isNull(ekklesiaCoreGrant.revokedAt),
                ...(input.sourceKind === undefined
                  ? []
                  : [eq(ekklesiaCoreGrant.sourceKind, input.sourceKind)]),
              ),
            )
            .limit(1);
          if (!target) {
            throw new AuthorizationStoreError("AUTHORIZATION_GRANT_NOT_FOUND");
          }

          await lockEquivalence(tx, target);

          // Revalidate the *addressed* grant now that the lock is held. Without this, a revocation
          // that paused between the read above and the lock could wake after its target was
          // revoked and a fresh equivalent created, and the tuple-matching update below would
          // silently revoke that replacement — reporting success for a grant the caller never
          // addressed, in an order no serial execution could produce, and losing the not-found
          // result an already-revoked target must return.
          //
          // The check is deliberately narrow and re-asserts every condition the pre-lock read
          // relied on: the exact addressed id, still unrevoked, still satisfying the caller's
          // explicit/relationship source guard, and still carrying the same equivalence tuple the
          // lock was taken for. It is a plain read, not `for update`: the tuple advisory lock
          // remains the one create/revoke serialization order, so no second lock ordering — and no
          // deadlock path — is introduced.
          const [addressed] = await tx
            .select({ id: ekklesiaCoreGrant.id })
            .from(ekklesiaCoreGrant)
            .where(
              and(
                eq(ekklesiaCoreGrant.id, input.grantId),
                isNull(ekklesiaCoreGrant.revokedAt),
                ...(input.sourceKind === undefined
                  ? []
                  : [eq(ekklesiaCoreGrant.sourceKind, input.sourceKind)]),
                equivalenceCondition(target),
              ),
            )
            .limit(1);
          if (!addressed) {
            throw new AuthorizationStoreError("AUTHORIZATION_GRANT_NOT_FOUND");
          }

          // Only now is the whole equivalence class cleared, which is what still removes genuine
          // legacy duplicates — including expired-but-unrevoked ones. Narrowing this update to the
          // addressed id instead would not have been a fix; it would have dropped that behavior.
          const rows = await tx
            .update(ekklesiaCoreGrant)
            .set({
              revokedAt: sql`now()`,
              revokedByActorId: actorId(input.actor),
              revokedByActorType: input.actor.type,
            })
            .where(and(equivalenceCondition(target), isNull(ekklesiaCoreGrant.revokedAt)))
            .returning();
          if (rows.length === 0) {
            // Unreachable while the lock is held, since the row confirmed above satisfies this
            // same predicate — kept as a fail-closed guard against a writer that bypasses the
            // lock, rather than silently auditing nothing.
            throw new AuthorizationStoreError("AUTHORIZATION_GRANT_NOT_FOUND");
          }

          await tx.insert(ekklesiaCoreAuthorizationAudit).values(
            rows.map((row) => ({
              action: "grant.revoke",
              actorId: actorId(input.actor),
              actorType: input.actor.type,
              grantId: row.id,
              permissionId: row.permissionId,
              scope: row.scope,
              subjectUserId: row.userId,
            })),
          );
        });
      } catch (error) {
        if (isAuthorizationStoreError(error)) {
          throw error;
        }
        throw new AuthorizationStoreError("AUTHORIZATION_GRANT_REVOKE_FAILED");
      }
    },

    hasActiveGrant: async (query: ActiveGrantQuery): Promise<boolean> => {
      try {
        const scopeCondition =
          query.scope === undefined
            ? isNull(ekklesiaCoreGrant.scope)
            : or(isNull(ekklesiaCoreGrant.scope), eq(ekklesiaCoreGrant.scope, query.scope));
        const rows = await db
          .select({ id: ekklesiaCoreGrant.id })
          .from(ekklesiaCoreGrant)
          .where(
            and(
              eq(ekklesiaCoreGrant.userId, query.userId),
              eq(ekklesiaCoreGrant.permissionId, query.permission),
              scopeCondition,
              isNull(ekklesiaCoreGrant.revokedAt),
              or(isNull(ekklesiaCoreGrant.expiresAt), gt(ekklesiaCoreGrant.expiresAt, query.now)),
            ),
          )
          .limit(1);
        return rows.length > 0;
      } catch {
        throw new AuthorizationStoreError("AUTHORIZATION_GRANT_LOOKUP_FAILED");
      }
    },

    /**
     * Deterministic keyset pagination over `created_at DESC, id DESC`. The boundary is a strict
     * row-value comparison on *both* columns, so grants sharing a `created_at` — which concurrent
     * creation makes routine — can neither be skipped nor repeated across pages. Offset pagination
     * would do exactly that under concurrent writes, which is why it is not used.
     *
     * At most `limit + 1` rows are read: the extra row only answers "is there another page" and is
     * never returned.
     */
    listGrantsForUser: async (
      userId: EkklesiaUserId,
      listOptions: ListGrantsForUserOptions,
    ): Promise<GrantPage> => {
      const limit = assertValidLimit(listOptions.limit);
      const after = listOptions.after;
      try {
        const rows = await db
          .select()
          .from(ekklesiaCoreGrant)
          .where(
            and(
              eq(ekklesiaCoreGrant.userId, userId),
              ...(listOptions.sourceKind === undefined
                ? []
                : [eq(ekklesiaCoreGrant.sourceKind, listOptions.sourceKind)]),
              ...(after === undefined
                ? []
                : [
                    sql`(${ekklesiaCoreGrant.createdAt}, ${ekklesiaCoreGrant.id}) < (${after.createdAt}::timestamptz, ${after.id}::uuid)`,
                  ]),
            ),
          )
          .orderBy(desc(ekklesiaCoreGrant.createdAt), desc(ekklesiaCoreGrant.id))
          .limit(limit + 1);

        const page = rows.slice(0, limit).map(toGrantRecord);
        const last = page.at(-1);
        return Object.freeze({
          grants: Object.freeze(page),
          ...(rows.length > limit && last !== undefined
            ? { nextBoundary: Object.freeze({ createdAt: last.createdAt, id: last.id }) }
            : {}),
        });
      } catch (error) {
        if (isAuthorizationStoreError(error)) {
          throw error;
        }
        throw new AuthorizationStoreError("AUTHORIZATION_GRANT_LIST_FAILED");
      }
    },

    grantPlatformOwner: async (
      userId: EkklesiaUserId,
      actor: AuthorizationActor,
    ): Promise<void> => {
      try {
        await db.transaction(async (tx) => {
          await tx
            .insert(ekklesiaCorePlatformOwner)
            .values({
              grantedByActorId: actorId(actor),
              grantedByActorType: actor.type,
              userId,
            })
            .onConflictDoUpdate({
              set: {
                grantedAt: sql`now()`,
                grantedByActorId: actorId(actor),
                grantedByActorType: actor.type,
                revokedAt: null,
                revokedByActorId: null,
                revokedByActorType: null,
              },
              target: ekklesiaCorePlatformOwner.userId,
            });

          await tx.insert(ekklesiaCoreAuthorizationAudit).values({
            action: "platform_owner.grant",
            actorId: actorId(actor),
            actorType: actor.type,
            subjectUserId: userId,
          });
        });
      } catch {
        throw new AuthorizationStoreError("AUTHORIZATION_PLATFORM_OWNER_MUTATION_FAILED");
      }
    },

    revokePlatformOwner: async (
      userId: EkklesiaUserId,
      actor: AuthorizationActor,
    ): Promise<void> => {
      try {
        await db.transaction(async (tx) => {
          const [row] = await tx
            .update(ekklesiaCorePlatformOwner)
            .set({
              revokedAt: sql`now()`,
              revokedByActorId: actorId(actor),
              revokedByActorType: actor.type,
            })
            .where(
              and(
                eq(ekklesiaCorePlatformOwner.userId, userId),
                isNull(ekklesiaCorePlatformOwner.revokedAt),
              ),
            )
            .returning();
          if (!row) {
            throw new AuthorizationStoreError("AUTHORIZATION_PLATFORM_OWNER_NOT_FOUND");
          }

          await tx.insert(ekklesiaCoreAuthorizationAudit).values({
            action: "platform_owner.revoke",
            actorId: actorId(actor),
            actorType: actor.type,
            subjectUserId: userId,
          });
        });
      } catch (error) {
        if (isAuthorizationStoreError(error)) {
          throw error;
        }
        throw new AuthorizationStoreError("AUTHORIZATION_PLATFORM_OWNER_MUTATION_FAILED");
      }
    },

    isPlatformOwner: async (userId: EkklesiaUserId): Promise<boolean> => {
      try {
        const rows = await db
          .select({ userId: ekklesiaCorePlatformOwner.userId })
          .from(ekklesiaCorePlatformOwner)
          .where(
            and(
              eq(ekklesiaCorePlatformOwner.userId, userId),
              isNull(ekklesiaCorePlatformOwner.revokedAt),
            ),
          )
          .limit(1);
        return rows.length > 0;
      } catch {
        throw new AuthorizationStoreError("AUTHORIZATION_PLATFORM_OWNER_LOOKUP_FAILED");
      }
    },
  });
}