import { ForbiddenException, NotFoundException } from "@nestjs/common";
import { and, desc, eq, inArray, isNull, lt, ne, or, sql } from "drizzle-orm";
import type Redis from "ioredis";
import type { Call, CallContact, CallKind, CallParticipant, CallRecordingStatus, CallTranscriptStatus } from "@slackwsh/contracts";
import { MAX_CALL_PARTICIPANTS } from "@slackwsh/contracts";
import { schema, withTenant, withUser } from "@slackwsh/data";
import { can } from "@slackwsh/core";
import { asEntityId, requireChannelMembership, requireWorkspaceMembership, type EntityIdInput } from "./channels";
import { activeWorkspaceMemberIds, fanoutToMembers } from "./member-fanout";

/**
 * How long a call may sit unanswered before it counts as missed.
 *
 * A ringing call has no natural end — the caller's browser may be closed
 * mid-ring, and nothing else would ever write the row. Rather than depend on
 * a worker that does not exist yet, reads sweep expired rows (see
 * sweepStaleCalls below), which keeps the invariant "no call is ringing
 * forever" true without a scheduler.
 */
export const CALL_RING_TIMEOUT_SECONDS = 45;

/**
 * How long an answered call may run with nobody sending a keepalive before it
 * is force-ended. This only catches calls whose participants all vanished
 * without a clean leave (a crashed tab, a killed laptop); a normal hangup
 * ends the call immediately via leaveCall.
 */
export const CALL_STALE_TIMEOUT_SECONDS = 12 * 60 * 60;

export interface StartCallInput {
  kind?: CallKind;
  inviteeUserIds: EntityIdInput[];
  channelId?: EntityIdInput | null;
  eventId?: EntityIdInput | null;
  title?: string | null;
}

export interface ListCallsFilters {
  active?: boolean;
  missed?: boolean;
  direction?: "incoming" | "outgoing";
  withUserId?: EntityIdInput;
  channelId?: EntityIdInput;
  limit?: number;
  before?: string;
}

type CallRow = typeof schema.calls.$inferSelect;
type ParticipantRow = typeof schema.callParticipants.$inferSelect;

async function requireCallInWorkspace(tx: any, callId: EntityIdInput, workspaceId: number): Promise<CallRow> {
  const id = asEntityId(callId);
  const [row] = await tx.select().from(schema.calls).where(eq(schema.calls.id, id)).limit(1);
  if (!row || row.workspaceId !== workspaceId) throw new NotFoundException("call not found");
  return row;
}

/** The channel a call optionally references must belong to the same workspace
 * — RLS alone only scopes by workspace_id, so caller-supplied id pairs still
 * need the explicit check every action in this codebase does. */
async function requireChannelInWorkspace(tx: any, channelId: EntityIdInput, workspaceId: number) {
  const chId = asEntityId(channelId);
  const [channel] = await tx.select().from(schema.channels).where(eq(schema.channels.id, chId)).limit(1);
  if (!channel || channel.workspaceId !== workspaceId) throw new NotFoundException("channel not found");
  return channel;
}

async function participantsFor(tx: any, callIds: number[]): Promise<Map<number, CallParticipant[]>> {
  const byCall = new Map<number, CallParticipant[]>();
  if (callIds.length === 0) return byCall;
  // One query for every call in the response — the alternative is an N+1
  // across a whole page of history.
  const rows: ParticipantRow[] = await tx
    .select()
    .from(schema.callParticipants)
    .where(inArray(schema.callParticipants.callId, callIds));
  for (const row of rows) {
    const list = byCall.get(row.callId) ?? [];
    list.push({
      userId: row.userId,
      state: row.state,
      invitedAt: row.invitedAt.toISOString(),
      joinedAt: row.joinedAt?.toISOString() ?? null,
      leftAt: row.leftAt?.toISOString() ?? null,
      seenAt: row.seenAt?.toISOString() ?? null,
    });
    byCall.set(row.callId, list);
  }
  return byCall;
}

/** Duration runs from the moment the call was answered, so an unanswered call
 * is 0 seconds and not "however long it rang". Null while still in progress —
 * a live call has no duration yet, and serving the elapsed time would make the
 * response stale the instant it was sent. */
function durationOf(row: CallRow): number | null {
  if (!row.endedAt) return null;
  if (!row.answeredAt) return 0;
  return Math.max(0, Math.round((row.endedAt.getTime() - row.answeredAt.getTime()) / 1000));
}

function recordingStatusOf(row: CallRow): CallRecordingStatus {
  const value = row.recordingStatus;
  if (value === "recording" || value === "processing" || value === "ready" || value === "failed") return value;
  return "idle";
}

function transcriptStatusOf(row: CallRow): CallTranscriptStatus {
  const value = row.transcriptStatus;
  if (value === "processing" || value === "ready" || value === "failed") return value;
  return "idle";
}

function toWireCall(row: CallRow, participants: CallParticipant[]): Call {
  return {
    id: row.id,
    workspaceId: row.workspaceId,
    kind: row.kind,
    status: row.status,
    channelId: row.channelId,
    eventId: row.eventId,
    title: row.title,
    startedBy: row.startedBy,
    startedAt: row.startedAt.toISOString(),
    answeredAt: row.answeredAt?.toISOString() ?? null,
    endedAt: row.endedAt?.toISOString() ?? null,
    durationSeconds: durationOf(row),
    participants: participants.slice().sort((a, b) => Number(a.userId) - Number(b.userId)),
    recordingStatus: recordingStatusOf(row),
    recordingObjectKey: row.recordingObjectKey ?? null,
    transcriptStatus: transcriptStatusOf(row),
    transcript: row.transcript ?? null,
    summary: row.summary ?? null,
  };
}

async function loadWireCall(tx: any, row: CallRow): Promise<Call> {
  const participants = await participantsFor(tx, [row.id]);
  return toWireCall(row, participants.get(row.id) ?? []);
}

/**
 * Ends calls nobody is going to come back to.
 *
 * Called from the read path rather than a scheduler, guarded by a short Redis
 * lock so a burst of concurrent readers performs the sweep once. It is
 * idempotent — the WHERE clauses only match rows still in a live status — so a
 * lost lock costs nothing but a skipped sweep, and the next read tries again.
 */
async function sweepStaleCalls(redis: Redis, workspaceId: number, userId: number): Promise<void> {
  const lock = await redis.set(`calls:sweep:${workspaceId}`, "1", "EX", 10, "NX").catch(() => null);
  if (!lock) return;

  const now = new Date();
  const ringCutoff = new Date(now.getTime() - CALL_RING_TIMEOUT_SECONDS * 1000);
  const staleCutoff = new Date(now.getTime() - CALL_STALE_TIMEOUT_SECONDS * 1000);

  const { retired, memberIds } = await withTenant({ workspaceId, userId }, async (tx) => {
    const expired: CallRow[] = await tx
      .select()
      .from(schema.calls)
      .where(
        and(
          eq(schema.calls.workspaceId, workspaceId),
          or(
            and(eq(schema.calls.status, "ringing"), lt(schema.calls.startedAt, ringCutoff)),
            and(eq(schema.calls.status, "active"), lt(schema.calls.startedAt, staleCutoff)),
          ),
        ),
      )
      .limit(50);

    const retired: Array<{ call: Call; missedBy: number[] }> = [];
    for (const row of expired) {
      // Captured before finalising, which is what rewrites these states.
      const reachable = await reachableUserIds(tx, row.id);
      const finalised = await finaliseCall(tx, row, now);
      retired.push({
        call: await loadWireCall(tx, finalised),
        missedBy: finalised.answeredAt ? [] : reachable.ringing,
      });
    }
    return { retired, memberIds: retired.length > 0 ? await activeWorkspaceMemberIds(tx, workspaceId) : [] };
  });

  // The sweep has to announce itself like every other way a call can end.
  // Without this the row went cold while both clients kept their last known
  // state: the caller sat on "Ringing…" with the microphone still open, the
  // invitee's ring card still offered Answer (which then failed with "this call
  // has ended"), and because notifyMissedCall was only wired into the explicit
  // hangup paths, the most common way to miss a call produced no notification.
  for (const { call, missedBy } of retired) {
    await fanoutToMembers(redis, memberIds, { type: "call:ended", payload: { call } });
    await notifyMissedCall(missedBy, call.startedBy, call);
  }
}

/**
 * Everyone who can still be reached on a call: on it, or being rung.
 *
 * Two of these is the threshold for a call still being a call, which is the one
 * rule leaving and declining share. Getting it wrong in either direction is
 * visible — too strict and a call collapses while someone is still dialling in,
 * too loose and one person is left alone in a room the workspace can see as
 * live. Note that counting *joined* alone is not enough: the initiator is
 * joined from the moment they dial, so "is anyone else joined" is true even
 * when nobody has picked up.
 */
async function reachableUserIds(tx: any, callId: number): Promise<{ joined: number[]; ringing: number[] }> {
  const rows = await tx
    .select({ userId: schema.callParticipants.userId, state: schema.callParticipants.state })
    .from(schema.callParticipants)
    .where(
      and(eq(schema.callParticipants.callId, callId), inArray(schema.callParticipants.state, ["joined", "ringing"])),
    );
  return {
    joined: rows.filter((r: { state: string }) => r.state === "joined").map((r: { userId: number }) => r.userId),
    ringing: rows.filter((r: { state: string }) => r.state === "ringing").map((r: { userId: number }) => r.userId),
  };
}

/**
 * Writes the terminal state of a call and its participants.
 *
 * A call that was never answered ends as `missed`, and every participant still
 * ringing is marked missed with it — that pairing is the whole basis of the
 * missed tab, so it lives in one place rather than at each call site that can
 * end a call (leave, explicit end, sweep).
 */
async function finaliseCall(tx: any, row: CallRow, at: Date): Promise<CallRow> {
  const status = row.answeredAt ? "ended" : "missed";
  const [updated] = await tx
    .update(schema.calls)
    .set({ status, endedAt: at })
    .where(eq(schema.calls.id, row.id))
    .returning();

  await tx
    .update(schema.callParticipants)
    .set({ state: "missed" })
    .where(and(eq(schema.callParticipants.callId, row.id), eq(schema.callParticipants.state, "ringing")));
  await tx
    .update(schema.callParticipants)
    .set({ state: "left", leftAt: at })
    .where(and(eq(schema.callParticipants.callId, row.id), eq(schema.callParticipants.state, "joined")));

  return (updated ?? row) as CallRow;
}

export async function startCall(
  redis: Redis,
  workspaceId: EntityIdInput,
  actorUserId: EntityIdInput,
  input: StartCallInput,
): Promise<Call> {
  const wsId = asEntityId(workspaceId);
  const userId = asEntityId(actorUserId);
  const channelId = input.channelId != null ? asEntityId(input.channelId) : null;
  const eventId = input.eventId != null ? asEntityId(input.eventId) : null;
  // The caller is always on the call; listing themselves as an invitee would
  // otherwise create a participant row that is both ringing and joined.
  const inviteeIds = [...new Set(input.inviteeUserIds.map((id) => asEntityId(id)))].filter((id) => id !== userId);

  const { call, wire, memberIds } = await withTenant({ workspaceId: wsId, userId }, async (tx) => {
    const membership = await requireWorkspaceMembership(tx, wsId, userId);
    if (!can({ userId, role: membership.role }, "call:start")) {
      throw new ForbiddenException("insufficient role to start calls");
    }
    if (inviteeIds.length === 0 && (input.kind ?? "audio") !== "connect") {
      throw new ForbiddenException("a call needs at least one other person");
    }
    // +1 for the caller — the contract caps the invitee list, this caps the
    // call, and they are not the same number.
    if (inviteeIds.length + 1 > MAX_CALL_PARTICIPANTS) {
      throw new ForbiddenException(`a call can hold at most ${MAX_CALL_PARTICIPANTS} people`);
    }
    if (channelId != null) await requireChannelInWorkspace(tx, channelId, wsId);
    if (eventId != null) {
      const [event] = await tx.select().from(schema.events).where(eq(schema.events.id, eventId)).limit(1);
      if (!event || event.workspaceId !== wsId) throw new NotFoundException("event not found");
    }
    for (const inviteeId of inviteeIds) await requireWorkspaceMembership(tx, wsId, inviteeId);

    const [inserted] = await tx
      .insert(schema.calls)
      .values({
        workspaceId: wsId,
        kind: input.kind ?? "audio",
        status: "ringing",
        channelId,
        eventId,
        title: input.title ?? null,
        startedBy: userId,
      })
      .returning();
    if (!inserted) throw new Error("call insert returned no row");

    const now = new Date();
    await tx.insert(schema.callParticipants).values([
      // The caller is on the call the moment it exists — they are not ringing
      // themselves, and their own history entry is never "missed".
      { callId: inserted.id, userId, state: "joined" as const, joinedAt: now, seenAt: now },
      ...inviteeIds.map((inviteeId) => ({ callId: inserted.id, userId: inviteeId, state: "ringing" as const })),
    ]);

    const wire = await loadWireCall(tx, inserted as CallRow);
    const memberIds = await activeWorkspaceMemberIds(tx, wsId);
    return { call: inserted as CallRow, wire, memberIds };
  });

  await fanoutToMembers(redis, memberIds, { type: "call:started", payload: { call: wire } });
  return wire;
}

export function livekitRoomName(workspaceId: number, callId: number): string {
  return `call-${workspaceId}-${callId}`;
}

/**
 * Join-or-start the persistent audio room for a channel (feature 8.1).
 *
 * There is at most one live Connect per channel (enforced by a unique index).
 * A collision from two people clicking at once joins the winner rather than
 * 500ing — the user asked to be in the room, not to be the one who created it.
 */
export async function startOrJoinConnect(
  redis: Redis,
  workspaceId: EntityIdInput,
  actorUserId: EntityIdInput,
  channelId: EntityIdInput,
): Promise<Call> {
  const wsId = asEntityId(workspaceId);
  const userId = asEntityId(actorUserId);
  const chId = asEntityId(channelId);

  const existing = await withTenant({ workspaceId: wsId, userId }, async (tx) => {
    await requireWorkspaceMembership(tx, wsId, userId);
    await requireChannelMembership(tx, chId, userId);
    const [row] = await tx
      .select()
      .from(schema.calls)
      .where(
        and(
          eq(schema.calls.workspaceId, wsId),
          eq(schema.calls.channelId, chId),
          eq(schema.calls.kind, "connect"),
          inArray(schema.calls.status, ["ringing", "active"]),
        ),
      )
      .limit(1);
    return (row as CallRow | undefined) ?? null;
  });

  if (existing) return joinCall(redis, wsId, userId, existing.id);

  try {
    const { wire, memberIds } = await withTenant({ workspaceId: wsId, userId }, async (tx) => {
      const membership = await requireWorkspaceMembership(tx, wsId, userId);
      if (!can({ userId, role: membership.role }, "call:start")) {
        throw new ForbiddenException("insufficient role to start calls");
      }
      const channel = await requireChannelInWorkspace(tx, chId, wsId);
      await requireChannelMembership(tx, chId, userId);

      const now = new Date();
      const [inserted] = await tx
        .insert(schema.calls)
        .values({
          workspaceId: wsId,
          kind: "connect",
          status: "active",
          channelId: chId,
          title: channel.name ? `Connect · ${channel.name}` : "Connect",
          startedBy: userId,
          answeredAt: now,
          livekitRoom: null,
        })
        .returning();
      if (!inserted) throw new Error("connect insert returned no row");

      await tx.insert(schema.callParticipants).values({
        callId: inserted.id,
        userId,
        state: "joined",
        joinedAt: now,
        seenAt: now,
      });

      const wire = await loadWireCall(tx, inserted as CallRow);
      const memberIds = await activeWorkspaceMemberIds(tx, wsId);
      return { wire, memberIds };
    });

    await fanoutToMembers(redis, memberIds, { type: "call:started", payload: { call: wire } });
    return wire;
  } catch (err) {
    const code = (err as { code?: string })?.code;
    if (code === "23505") {
      const raced = await withTenant({ workspaceId: wsId, userId }, async (tx) => {
        const [row] = await tx
          .select()
          .from(schema.calls)
          .where(
            and(
              eq(schema.calls.workspaceId, wsId),
              eq(schema.calls.channelId, chId),
              eq(schema.calls.kind, "connect"),
              inArray(schema.calls.status, ["ringing", "active"]),
            ),
          )
          .limit(1);
        return (row as CallRow | undefined) ?? null;
      });
      if (raced) return joinCall(redis, wsId, userId, raced.id);
    }
    throw err;
  }
}

export async function bindLivekitRoom(
  workspaceId: EntityIdInput,
  actorUserId: EntityIdInput,
  callId: EntityIdInput,
  roomName: string,
): Promise<Call> {
  const wsId = asEntityId(workspaceId);
  const userId = asEntityId(actorUserId);
  return withTenant({ workspaceId: wsId, userId }, async (tx) => {
    const row = await requireCallInWorkspace(tx, callId, wsId);
    if (row.livekitRoom === roomName) return loadWireCall(tx, row);
    const [updated] = await tx
      .update(schema.calls)
      .set({ livekitRoom: roomName })
      .where(eq(schema.calls.id, row.id))
      .returning();
    return loadWireCall(tx, (updated ?? row) as CallRow);
  });
}

export async function setCallRecording(
  workspaceId: EntityIdInput,
  actorUserId: EntityIdInput,
  callId: EntityIdInput,
  patch: {
    recordingStatus?: CallRecordingStatus;
    recordingEgressId?: string | null;
    recordingObjectKey?: string | null;
  },
): Promise<Call> {
  const wsId = asEntityId(workspaceId);
  const userId = asEntityId(actorUserId);
  return withTenant({ workspaceId: wsId, userId }, async (tx) => {
    const row = await requireCallInWorkspace(tx, callId, wsId);
    const [updated] = await tx
      .update(schema.calls)
      .set({
        recordingStatus: patch.recordingStatus ?? row.recordingStatus,
        recordingEgressId: patch.recordingEgressId === undefined ? row.recordingEgressId : patch.recordingEgressId,
        recordingObjectKey: patch.recordingObjectKey === undefined ? row.recordingObjectKey : patch.recordingObjectKey,
      })
      .where(eq(schema.calls.id, row.id))
      .returning();
    return loadWireCall(tx, (updated ?? row) as CallRow);
  });
}

export async function setCallTranscript(
  workspaceId: EntityIdInput,
  actorUserId: EntityIdInput,
  callId: EntityIdInput,
  patch: {
    transcriptStatus?: CallTranscriptStatus;
    transcript?: string | null;
    summary?: string | null;
  },
): Promise<Call> {
  const wsId = asEntityId(workspaceId);
  const userId = asEntityId(actorUserId);
  return withTenant({ workspaceId: wsId, userId }, async (tx) => {
    const row = await requireCallInWorkspace(tx, callId, wsId);
    const [updated] = await tx
      .update(schema.calls)
      .set({
        transcriptStatus: patch.transcriptStatus ?? row.transcriptStatus,
        transcript: patch.transcript === undefined ? row.transcript : patch.transcript,
        summary: patch.summary === undefined ? row.summary : patch.summary,
      })
      .where(eq(schema.calls.id, row.id))
      .returning();
    return loadWireCall(tx, (updated ?? row) as CallRow);
  });
}

/** Resolve an inbound PSTN number to a workspace. Env mapping is the
 * Phase 6 path (one number per deployment); the workspace_phone_numbers
 * table is the multi-tenant follow-up. */
export function workspaceIdForInboundPstn(e164: string): number | null {
  const configured = (process.env.TWILIO_PSTN_NUMBER ?? "").replace(/\s/g, "");
  const ws = Number(process.env.TWILIO_PSTN_WORKSPACE_ID);
  if (!configured || configured !== e164.replace(/\s/g, "")) return null;
  if (!Number.isInteger(ws) || ws <= 0) return null;
  return ws;
}

/**
 * Reads call history.
 *
 * Scoped to calls the actor took part in, which is stricter than the policy
 * table can express: `call:read` grants access to the feature, not to other
 * people's call logs. The one exception is the live-calls read, where a call
 * attached to a channel the actor belongs to is joinable by them and therefore
 * has to be visible — the same "a Connect in your room is yours to join" rule
 * the chat UI implies.
 */
export async function listCalls(
  redis: Redis,
  workspaceId: EntityIdInput,
  actorUserId: EntityIdInput,
  filters: ListCallsFilters = {},
): Promise<Call[]> {
  const wsId = asEntityId(workspaceId);
  const userId = asEntityId(actorUserId);
  const limit = Math.min(Math.max(filters.limit ?? 50, 1), 100);

  // Before reading, retire anything the world moved on from — otherwise a
  // caller who closed their laptop mid-ring leaves a call ringing forever in
  // everyone else's active strip.
  await sweepStaleCalls(redis, wsId, userId);

  return withTenant({ workspaceId: wsId, userId }, async (tx) => {
    const membership = await requireWorkspaceMembership(tx, wsId, userId);
    if (!can({ userId, role: membership.role }, "call:read")) {
      throw new ForbiddenException("insufficient role to read calls");
    }

    const mine = tx
      .select({ id: schema.callParticipants.callId })
      .from(schema.callParticipants)
      .where(eq(schema.callParticipants.userId, userId));

    const channelId = filters.channelId != null ? asEntityId(filters.channelId) : null;
    if (channelId != null) await requireChannelInWorkspace(tx, channelId, wsId);

    const liveOnChannel =
      channelId != null
        ? and(
            eq(schema.calls.channelId, channelId),
            inArray(schema.calls.status, ["ringing", "active"]),
          )
        : undefined;

    const visible = filters.active
      ? liveOnChannel
        ? or(
            inArray(
              schema.calls.id,
              tx
                .select({ id: schema.callParticipants.callId })
                .from(schema.callParticipants)
                .where(
                  and(
                    eq(schema.callParticipants.userId, userId),
                    inArray(schema.callParticipants.state, ["ringing", "joined"]),
                  ),
                ),
            ),
            liveOnChannel,
          )
        : inArray(
            schema.calls.id,
            tx
              .select({ id: schema.callParticipants.callId })
              .from(schema.callParticipants)
              .where(
                and(
                  eq(schema.callParticipants.userId, userId),
                  inArray(schema.callParticipants.state, ["ringing", "joined"]),
                ),
              ),
          )
      : liveOnChannel
        ? or(inArray(schema.calls.id, mine), liveOnChannel)
        : inArray(schema.calls.id, mine);

    const where = [eq(schema.calls.workspaceId, wsId), visible];
    if (filters.active) where.push(inArray(schema.calls.status, ["ringing", "active"]));
    if (filters.direction === "outgoing") where.push(eq(schema.calls.startedBy, userId));
    if (filters.direction === "incoming") where.push(ne(schema.calls.startedBy, userId));
    if (filters.before) where.push(lt(schema.calls.startedAt, new Date(filters.before)));
    if (filters.missed) {
      // Missed is a property of *this* reader's participant row: the same call
      // was answered by someone else.
      where.push(
        inArray(
          schema.calls.id,
          tx
            .select({ id: schema.callParticipants.callId })
            .from(schema.callParticipants)
            .where(
              and(eq(schema.callParticipants.userId, userId), eq(schema.callParticipants.state, "missed")),
            ),
        ),
      );
    }
    if (filters.withUserId != null) {
      const withId = asEntityId(filters.withUserId);
      where.push(
        inArray(
          schema.calls.id,
          tx
            .select({ id: schema.callParticipants.callId })
            .from(schema.callParticipants)
            .where(eq(schema.callParticipants.userId, withId)),
        ),
      );
    }
    if (filters.channelId != null) {
      where.push(eq(schema.calls.channelId, asEntityId(filters.channelId)));
    }

    const rows: CallRow[] = await tx
      .select()
      .from(schema.calls)
      .where(and(...where))
      .orderBy(desc(schema.calls.startedAt))
      .limit(limit);

    const participants = await participantsFor(
      tx,
      rows.map((row) => row.id),
    );
    return rows.map((row) => toWireCall(row, participants.get(row.id) ?? []));
  });
}

export async function getCall(
  workspaceId: EntityIdInput,
  actorUserId: EntityIdInput,
  callId: EntityIdInput,
): Promise<Call> {
  const wsId = asEntityId(workspaceId);
  const userId = asEntityId(actorUserId);
  return withTenant({ workspaceId: wsId, userId }, async (tx) => {
    const membership = await requireWorkspaceMembership(tx, wsId, userId);
    if (!can({ userId, role: membership.role }, "call:read")) {
      throw new ForbiddenException("insufficient role to read calls");
    }
    const row = await requireCallInWorkspace(tx, callId, wsId);
    return loadWireCall(tx, row);
  });
}

/** Whether a user may exchange media with a call — the authorisation the
 * gateway's signalling relay needs. Only someone currently on the call can
 * signal, so a declined or departed participant cannot keep negotiating. */
export async function isCallParticipant(
  workspaceId: EntityIdInput,
  callId: EntityIdInput,
  userId: EntityIdInput,
): Promise<boolean> {
  const wsId = asEntityId(workspaceId);
  const uid = asEntityId(userId);
  const cid = asEntityId(callId);
  return withTenant({ workspaceId: wsId, userId: uid }, async (tx) => {
    const [row] = await tx
      .select({ state: schema.callParticipants.state })
      .from(schema.callParticipants)
      .innerJoin(schema.calls, eq(schema.calls.id, schema.callParticipants.callId))
      .where(
        and(
          eq(schema.callParticipants.callId, cid),
          eq(schema.callParticipants.userId, uid),
          eq(schema.calls.workspaceId, wsId),
          inArray(schema.calls.status, ["ringing", "active"]),
        ),
      )
      .limit(1);
    return row?.state === "joined" || row?.state === "ringing";
  });
}

/**
 * Answers a call, or walks into a channel call uninvited.
 *
 * The second case is why this can add a participant row rather than only
 * update one: a call attached to a room is open to that room's members, so
 * joining one is a legitimate first contact with the call. A call with no
 * channel is closed — only the people who were rung may answer.
 */
export async function joinCall(
  redis: Redis,
  workspaceId: EntityIdInput,
  actorUserId: EntityIdInput,
  callId: EntityIdInput,
): Promise<Call> {
  const wsId = asEntityId(workspaceId);
  const userId = asEntityId(actorUserId);

  const { wire, memberIds } = await withTenant({ workspaceId: wsId, userId }, async (tx) => {
    const membership = await requireWorkspaceMembership(tx, wsId, userId);
    if (!can({ userId, role: membership.role }, "call:join")) {
      throw new ForbiddenException("insufficient role to join calls");
    }
    const row = await requireCallInWorkspace(tx, callId, wsId);
    if (row.status === "ended" || row.status === "missed") throw new ForbiddenException("this call has ended");

    const [existing] = await tx
      .select()
      .from(schema.callParticipants)
      .where(and(eq(schema.callParticipants.callId, row.id), eq(schema.callParticipants.userId, userId)))
      .limit(1);

    if (!existing) {
      if (row.channelId == null) throw new ForbiddenException("you were not invited to this call");
      const [inChannel] = await tx
        .select({ userId: schema.channelMembers.userId })
        .from(schema.channelMembers)
        .where(and(eq(schema.channelMembers.channelId, row.channelId), eq(schema.channelMembers.userId, userId)))
        .limit(1);
      if (!inChannel) throw new ForbiddenException("you were not invited to this call");
    }

    const now = new Date();
    const joined = await tx
      .select({ userId: schema.callParticipants.userId })
      .from(schema.callParticipants)
      .where(and(eq(schema.callParticipants.callId, row.id), eq(schema.callParticipants.state, "joined")));
    const alreadyOn = joined.some((p: { userId: number }) => p.userId === userId);
    const headcount = joined.length + (alreadyOn ? 0 : 1);
    if (!alreadyOn && headcount > MAX_CALL_PARTICIPANTS) {
      throw new ForbiddenException(`a call can hold at most ${MAX_CALL_PARTICIPANTS} people`);
    }

    if (existing) {
      await tx
        .update(schema.callParticipants)
        .set({ state: "joined", joinedAt: existing.joinedAt ?? now, leftAt: null, seenAt: now })
        .where(and(eq(schema.callParticipants.callId, row.id), eq(schema.callParticipants.userId, userId)));
    } else {
      await tx
        .insert(schema.callParticipants)
        .values({ callId: row.id, userId, state: "joined", joinedAt: now, seenAt: now });
    }

    // Two people on the call at once is what "answered" means, and the first
    // time it happens is what duration is measured from.
    const patch: Record<string, unknown> = {};
    if (row.status === "ringing" && headcount >= 2) patch.status = "active";
    if (row.answeredAt == null && headcount >= 2) patch.answeredAt = now;

    let current = row;
    if (Object.keys(patch).length > 0) {
      const [updated] = await tx.update(schema.calls).set(patch).where(eq(schema.calls.id, row.id)).returning();
      if (updated) current = updated as CallRow;
    }

    const wire = await loadWireCall(tx, current);
    const memberIds = await activeWorkspaceMemberIds(tx, wsId);
    return { wire, memberIds };
  });

  await fanoutToMembers(redis, memberIds, { type: "call:updated", payload: { call: wire } });
  return wire;
}

/** Turns down a call. Deliberately takes no target user — you may only decline
 * for yourself, which the role table can't express (it has no notion of
 * "subject equals actor"), so it is enforced here by construction. */
export async function declineCall(
  redis: Redis,
  workspaceId: EntityIdInput,
  actorUserId: EntityIdInput,
  callId: EntityIdInput,
): Promise<Call> {
  const wsId = asEntityId(workspaceId);
  const userId = asEntityId(actorUserId);

  const { wire, memberIds } = await withTenant({ workspaceId: wsId, userId }, async (tx) => {
    const membership = await requireWorkspaceMembership(tx, wsId, userId);
    if (!can({ userId, role: membership.role }, "call:join")) {
      throw new ForbiddenException("insufficient role to answer calls");
    }
    const row = await requireCallInWorkspace(tx, callId, wsId);
    const [existing] = await tx
      .select()
      .from(schema.callParticipants)
      .where(and(eq(schema.callParticipants.callId, row.id), eq(schema.callParticipants.userId, userId)))
      .limit(1);
    if (!existing) throw new NotFoundException("you were not invited to this call");

    const now = new Date();
    await tx
      .update(schema.callParticipants)
      .set({ state: "declined", seenAt: now })
      .where(and(eq(schema.callParticipants.callId, row.id), eq(schema.callParticipants.userId, userId)));

    // A one-to-one call that has been declined is over: only the caller is
    // left, and leaving it `ringing` would keep them staring at a dialog until
    // the sweep timeout. A group call, where someone else is still ringing or
    // already talking, carries on without the decliner.
    const reachable = await reachableUserIds(tx, row.id);
    const viable = reachable.joined.length + reachable.ringing.length >= 2;
    const live = row.status === "ringing" || row.status === "active";
    const current = live && !viable ? await finaliseCall(tx, row, now) : row;

    const wire = await loadWireCall(tx, current);
    const memberIds = await activeWorkspaceMemberIds(tx, wsId);
    return { wire, memberIds };
  });

  await fanoutToMembers(redis, memberIds, {
    type: wire.endedAt ? "call:ended" : "call:updated",
    payload: { call: wire },
  });
  return wire;
}

/**
 * Hangs up for one person, and ends the call if they were the last one on it.
 *
 * "Last one" counts people still ringing too: a caller who gives up while the
 * phone is still ringing has ended the call, not left the others to talk among
 * themselves.
 */
export async function leaveCall(
  redis: Redis,
  workspaceId: EntityIdInput,
  actorUserId: EntityIdInput,
  callId: EntityIdInput,
): Promise<Call> {
  const wsId = asEntityId(workspaceId);
  const userId = asEntityId(actorUserId);

  const { wire, memberIds, missedBy, starter } = await withTenant({ workspaceId: wsId, userId }, async (tx) => {
    const membership = await requireWorkspaceMembership(tx, wsId, userId);
    if (!can({ userId, role: membership.role }, "call:end")) {
      throw new ForbiddenException("insufficient role to end calls");
    }
    const row = await requireCallInWorkspace(tx, callId, wsId);
    const now = new Date();

    await tx
      .update(schema.callParticipants)
      .set({ state: "left", leftAt: now, seenAt: now })
      .where(
        and(
          eq(schema.callParticipants.callId, row.id),
          eq(schema.callParticipants.userId, userId),
          inArray(schema.callParticipants.state, ["joined", "ringing"]),
        ),
      );

    // One person alone is not a call, whether the others hung up or never
    // picked up in the first place — the same threshold declineCall uses.
    const reachable = await reachableUserIds(tx, row.id);
    const ends =
      row.status !== "ended" &&
      row.status !== "missed" &&
      reachable.joined.length + reachable.ringing.length < 2;
    const current = ends ? await finaliseCall(tx, row, now) : row;

    const wire = await loadWireCall(tx, current);
    const memberIds = await activeWorkspaceMemberIds(tx, wsId);
    return {
      wire,
      memberIds,
      missedBy: ends && !current.answeredAt ? reachable.ringing : [],
      starter: current.startedBy,
    };
  });

  await fanoutToMembers(redis, memberIds, {
    type: wire.endedAt ? "call:ended" : "call:updated",
    payload: { call: wire },
  });
  await notifyMissedCall(missedBy, starter, wire);
  return wire;
}

/** Ends the call for everyone. Distinct from leaveCall, which only ends it as
 * a side effect of the room emptying — this is the deliberate "hang up on all
 * of us", available to anyone on the call. */
export async function endCall(
  redis: Redis,
  workspaceId: EntityIdInput,
  actorUserId: EntityIdInput,
  callId: EntityIdInput,
): Promise<Call> {
  const wsId = asEntityId(workspaceId);
  const userId = asEntityId(actorUserId);

  const { wire, memberIds, missedBy, starter } = await withTenant({ workspaceId: wsId, userId }, async (tx) => {
    const membership = await requireWorkspaceMembership(tx, wsId, userId);
    if (!can({ userId, role: membership.role }, "call:end")) {
      throw new ForbiddenException("insufficient role to end calls");
    }
    const row = await requireCallInWorkspace(tx, callId, wsId);
    const [me] = await tx
      .select({ state: schema.callParticipants.state })
      .from(schema.callParticipants)
      .where(and(eq(schema.callParticipants.callId, row.id), eq(schema.callParticipants.userId, userId)))
      .limit(1);
    if (!me) throw new ForbiddenException("you are not on this call");

    const now = new Date();
    const ringing = await tx
      .select({ userId: schema.callParticipants.userId })
      .from(schema.callParticipants)
      .where(and(eq(schema.callParticipants.callId, row.id), eq(schema.callParticipants.state, "ringing")));
    const current =
      row.status === "ended" || row.status === "missed" ? row : await finaliseCall(tx, row, now);

    const wire = await loadWireCall(tx, current);
    const memberIds = await activeWorkspaceMemberIds(tx, wsId);
    return {
      wire,
      memberIds,
      missedBy: current.answeredAt ? [] : ringing.map((p: { userId: number }) => p.userId),
      starter: current.startedBy,
    };
  });

  await fanoutToMembers(redis, memberIds, { type: "call:ended", payload: { call: wire } });
  await notifyMissedCall(missedBy, starter, wire);
  return wire;
}

/** Adds people to a call in progress. Additive, unlike inviteToEvent's
 * replace-wholesale: the people already on the call are not the caller's rows
 * to rewrite. */
export async function inviteToCall(
  redis: Redis,
  workspaceId: EntityIdInput,
  actorUserId: EntityIdInput,
  callId: EntityIdInput,
  inviteeUserIds: EntityIdInput[],
): Promise<Call> {
  const wsId = asEntityId(workspaceId);
  const userId = asEntityId(actorUserId);
  const inviteeIds = [...new Set(inviteeUserIds.map((id) => asEntityId(id)))];

  const { wire, memberIds } = await withTenant({ workspaceId: wsId, userId }, async (tx) => {
    const membership = await requireWorkspaceMembership(tx, wsId, userId);
    if (!can({ userId, role: membership.role }, "call:start")) {
      throw new ForbiddenException("insufficient role to invite to calls");
    }
    const row = await requireCallInWorkspace(tx, callId, wsId);
    if (row.status === "ended" || row.status === "missed") throw new ForbiddenException("this call has ended");
    for (const inviteeId of inviteeIds) await requireWorkspaceMembership(tx, wsId, inviteeId);

    const existing = await tx
      .select({ userId: schema.callParticipants.userId })
      .from(schema.callParticipants)
      .where(eq(schema.callParticipants.callId, row.id));
    const known = new Set(existing.map((p: { userId: number }) => p.userId));
    const toAdd = inviteeIds.filter((id) => !known.has(id));
    if (known.size + toAdd.length > MAX_CALL_PARTICIPANTS) {
      throw new ForbiddenException(`a call can hold at most ${MAX_CALL_PARTICIPANTS} people`);
    }
    if (toAdd.length > 0) {
      await tx
        .insert(schema.callParticipants)
        .values(toAdd.map((inviteeId) => ({ callId: row.id, userId: inviteeId, state: "ringing" as const })));
    }

    const wire = await loadWireCall(tx, row);
    const memberIds = await activeWorkspaceMemberIds(tx, wsId);
    return { wire, memberIds };
  });

  await fanoutToMembers(redis, memberIds, { type: "call:updated", payload: { call: wire } });
  return wire;
}

/** Clears the missed-call badge. Only ever touches the actor's own rows, so
 * there is no target user to authorise. */
export async function markCallsSeen(
  workspaceId: EntityIdInput,
  actorUserId: EntityIdInput,
  callIds?: EntityIdInput[],
): Promise<{ seen: number }> {
  const wsId = asEntityId(workspaceId);
  const userId = asEntityId(actorUserId);
  const ids = callIds?.map((id) => asEntityId(id));

  return withTenant({ workspaceId: wsId, userId }, async (tx) => {
    const membership = await requireWorkspaceMembership(tx, wsId, userId);
    if (!can({ userId, role: membership.role }, "call:read")) {
      throw new ForbiddenException("insufficient role to read calls");
    }

    const scope = [
      eq(schema.callParticipants.userId, userId),
      isNull(schema.callParticipants.seenAt),
      // RLS scopes call_participants through its call, but the workspace of a
      // *specific* id pair still has to be checked explicitly.
      inArray(
        schema.callParticipants.callId,
        tx.select({ id: schema.calls.id }).from(schema.calls).where(eq(schema.calls.workspaceId, wsId)),
      ),
    ];
    if (ids && ids.length > 0) scope.push(inArray(schema.callParticipants.callId, ids));

    const updated = await tx
      .update(schema.callParticipants)
      .set({ seenAt: new Date() })
      .where(and(...scope))
      .returning({ callId: schema.callParticipants.callId });
    return { seen: updated.length };
  });
}

/** How many calls the actor has never acknowledged — the nav badge's number. */
export async function unseenMissedCallCount(
  workspaceId: EntityIdInput,
  actorUserId: EntityIdInput,
): Promise<number> {
  const wsId = asEntityId(workspaceId);
  const userId = asEntityId(actorUserId);
  return withTenant({ workspaceId: wsId, userId }, async (tx) => {
    await requireWorkspaceMembership(tx, wsId, userId);
    const rows = await tx
      .select({ callId: schema.callParticipants.callId })
      .from(schema.callParticipants)
      .innerJoin(schema.calls, eq(schema.calls.id, schema.callParticipants.callId))
      .where(
        and(
          eq(schema.callParticipants.userId, userId),
          eq(schema.callParticipants.state, "missed"),
          isNull(schema.callParticipants.seenAt),
          eq(schema.calls.workspaceId, wsId),
        ),
      );
    return rows.length;
  });
}

/**
 * The "people to call" ranking: everyone the actor has actually called,
 * with how often and how recently.
 *
 * Aggregated in SQL over a self-join of call_participants (my calls, then
 * everyone else on them) rather than by paging history client-side — the
 * ranking needs every call ever, which is exactly the read a page size can't
 * give you. Members with no shared call history are absent from the result;
 * the client unions this with the member roster so a new user still sees
 * someone to call.
 */
export async function callContacts(
  workspaceId: EntityIdInput,
  actorUserId: EntityIdInput,
): Promise<CallContact[]> {
  const wsId = asEntityId(workspaceId);
  const userId = asEntityId(actorUserId);

  return withTenant({ workspaceId: wsId, userId }, async (tx) => {
    const membership = await requireWorkspaceMembership(tx, wsId, userId);
    if (!can({ userId, role: membership.role }, "call:read")) {
      throw new ForbiddenException("insufficient role to read calls");
    }

    const mine = tx
      .select({ id: schema.callParticipants.callId })
      .from(schema.callParticipants)
      .where(eq(schema.callParticipants.userId, userId));

    const rows = await tx
      .select({
        userId: schema.callParticipants.userId,
        callCount: sql<number>`count(*)::int`,
        lastCallAt: sql<Date | null>`max(${schema.calls.startedAt})`,
      })
      .from(schema.callParticipants)
      .innerJoin(schema.calls, eq(schema.calls.id, schema.callParticipants.callId))
      .where(
        and(
          eq(schema.calls.workspaceId, wsId),
          inArray(schema.callParticipants.callId, mine),
          ne(schema.callParticipants.userId, userId),
        ),
      )
      .groupBy(schema.callParticipants.userId);

    return rows
      .map((row: { userId: number; callCount: number; lastCallAt: Date | string | null }) => ({
        userId: row.userId,
        callCount: Number(row.callCount),
        lastCallAt: row.lastCallAt ? new Date(row.lastCallAt).toISOString() : null,
      }))
      .sort((a: CallContact, b: CallContact) => {
        if (b.callCount !== a.callCount) return b.callCount - a.callCount;
        return (b.lastCallAt ?? "").localeCompare(a.lastCallAt ?? "");
      });
  });
}

/** Same direct-insert-per-user pattern as notifyInvitees in calendar.ts,
 * since apps/worker's notification consumer is still a logging stub. A missed
 * call is the one call event worth a persistent notification — the others all
 * happened while you were looking at them. */
async function notifyMissedCall(missedByUserIds: number[], callerUserId: number, call: Call) {
  for (const userId of missedByUserIds) {
    if (userId === callerUserId) continue;
    await withUser(userId, (tx) =>
      tx.insert(schema.notifications).values({
        userId,
        type: "missed_call",
        title: "Missed call",
        body: call.kind === "video" ? "You missed a video call" : "You missed a call",
        payload: {
          callId: call.id,
          workspaceId: call.workspaceId,
          kind: call.kind,
          startedAt: call.startedAt,
          fromUserId: callerUserId,
        },
      }),
    );
  }
}