import { all, chunks, first, placeholders, run } from "./db";
import { accountDb, repoDb, stmt } from "./scoped-db";
import { forgetLocator } from "./scoped-db/locators";
import { allUserIds, forEachScope } from "./account-scope";
import { repoReadableSql } from "./repos";
import { invalidateUnread } from "./user-inbox";
import { env } from "cloudflare:workers";
import { addCounter } from "./counters";
import { REPO_STARS } from "./counter-keys";

/**
 * Stars, watches, thread subscriptions, assignments, review requests and notifications feed the dashboard and
 * inbox; rows held by users who can no longer read the repository must go when access is revoked
 * (collaborator removal, org membership removal, public -> private), otherwise private activity keeps
 * streaming to them. Stars and watches are D1 edges; subscriptions, assignments and review requests live in
 * each repository's RepoState, notifications in each recipient's AccountState.
 */
const CANNOT_READ = `NOT EXISTS (SELECT 1 FROM repos r WHERE r.id = t.repo_id AND ${repoReadableSql("r", "t.user_id")})
  AND NOT EXISTS (SELECT 1 FROM users u_ WHERE u_.account_id = t.user_id AND u_.is_admin = 1)`;

/** Of `userIds`, those who can no longer read `repoId` (a deleted repository is unreadable to everyone). */
async function unreadableBy(repoId: string, userIds: string[]): Promise<string[]> {
  const out: string[] = [];
  for (const ids of chunks([...new Set(userIds)])) {
    const rows = await all<{ user_id: string }>(
      `SELECT t.user_id FROM (SELECT a.id AS user_id, ? AS repo_id FROM accounts a WHERE a.id IN (${placeholders(ids.length)})) t
       WHERE ${CANNOT_READ}`,
      repoId,
      ...ids,
    );
    out.push(...rows.map((row) => row.user_id));
  }
  return out;
}

const REPO_EDGES = ["thread_subscriptions", "issue_assignees", "review_requests"] as const;

async function purgeRepoEdges(repoId: string, userId: string | null): Promise<void> {
  const db = repoDb(repoId);
  const userFilter = userId === null ? "" : " AND x.user_id = ?";
  const candidates = new Set<string>();
  for (const table of REPO_EDGES) {
    const rows = await db.all<{ user_id: string }>(
      `SELECT DISTINCT x.user_id FROM ${table} x JOIN issues i ON i.id = x.issue_id WHERE i.repo_id = ?${userFilter}`,
      ...(userId === null ? [repoId] : [repoId, userId]),
    );
    for (const row of rows) candidates.add(row.user_id);
  }
  const revoked = await unreadableBy(repoId, [...candidates]);
  if (revoked.length === 0) return;
  await db.batch(
    REPO_EDGES.flatMap((table) =>
      chunks(revoked).map((ids) =>
        stmt(
          `DELETE FROM ${table} WHERE user_id IN (${placeholders(ids.length)})
           AND issue_id IN (SELECT id FROM issues WHERE repo_id = ?)`,
          ...ids,
          repoId,
        ),
      ),
    ),
  );
}

/** Removes notifications on `repoIds` from recipients who lost read access to them. */
async function purgeNotifications(repoIds: string[], userId: string | null): Promise<void> {
  if (repoIds.length === 0) return;
  const recipients = userId === null ? await allUserIds() : [userId];
  const touched: string[] = [];
  await forEachScope(recipients, async (recipient) => {
    const db = accountDb(recipient);
    let removed = 0;
    for (const ids of chunks(repoIds)) {
      const held = await db.all<{ repo_id: string }>(
        `SELECT DISTINCT repo_id FROM notifications WHERE user_id = ? AND repo_id IN (${placeholders(ids.length)})`,
        recipient,
        ...ids,
      );
      for (const { repo_id } of held) {
        if ((await unreadableBy(repo_id, [recipient])).length === 0) continue;
        const result = await db.run("DELETE FROM notifications WHERE user_id = ? AND repo_id = ?", recipient, repo_id);
        removed += result.meta.changes;
      }
    }
    if (removed > 0) touched.push(recipient);
  });
  if (touched.length > 0) await invalidateUnread(touched);
}

/** Removed stars must move the repo-counts counter, or the next flush would write the stale total back. */
async function purgeStars(repoFilter: string, userFilter: string, params: string[]): Promise<void> {
  const removed = await all<{ repo_id: string }>(
    `DELETE FROM stars WHERE (user_id, repo_id) IN (SELECT t.user_id, t.repo_id FROM stars t WHERE ${repoFilter} ${userFilter} AND ${CANNOT_READ})
     RETURNING repo_id`,
    ...params,
  );
  const perRepo = new Map<string, number>();
  for (const row of removed) perRepo.set(row.repo_id, (perRepo.get(row.repo_id) ?? 0) + 1);
  await Promise.all(
    [...perRepo].map(([repoId, count]) => addCounter("repo-counts", repoId, REPO_STARS, [-count, 0, 0])),
  );
}

async function purge(repoIds: string[], repoFilter: string, userId: string | null, ...params: string[]): Promise<void> {
  const userFilter = userId === null ? "" : "AND t.user_id = ?2";
  await purgeStars(repoFilter, userFilter, params);
  await run(
    `DELETE FROM watches WHERE (user_id, repo_id) IN (SELECT t.user_id, t.repo_id FROM watches t WHERE ${repoFilter} ${userFilter} AND ${CANNOT_READ})`,
    ...params,
  );
  for (const repoId of repoIds) await purgeRepoEdges(repoId, userId);
  await purgeNotifications(repoIds, userId);
}

/** Still an admin of the repository another way: its owner, an owner of the owning org, or an admin collaborator. */
const STILL_ADMIN = `SELECT 1 AS ok FROM repos r WHERE r.id = ?1 AND (r.owner_id = ?2
  OR EXISTS (SELECT 1 FROM collaborators c WHERE c.repo_id = r.id AND c.user_id = ?2 AND c.permission = 'admin')
  OR EXISTS (SELECT 1 FROM org_members m WHERE m.org_id = r.owner_id AND m.user_id = ?2 AND m.role = 'owner'))`;

/**
 * Deploy tokens follow admin standing (AUTH-P1/P9): when someone loses admin on a repository (collaborator
 * removed or downgraded, org owner removed or demoted) the deploy tokens they created there are revoked,
 * unless they still hold admin another way. D1 did this in the collaborators/org_members triggers; the
 * tokens now live in each repository's RepoState.
 */
export async function revokeDeployTokensCreatedBy(userId: string, repoIds: string[]): Promise<void> {
  for (const repoId of repoIds) {
    if (await first<{ ok: number }>(STILL_ADMIN, repoId, userId)) continue;
    const revoked = await repoDb(repoId).all<{ token_hash: string }>(
      "DELETE FROM deploy_tokens WHERE repo_id = ? AND created_by = ? RETURNING token_hash",
      repoId,
      userId,
    );
    await Promise.all(revoked.map((row) => forgetLocator("deploy_token", row.token_hash)));
  }
}

/** MCP sessions cache granted access for ACCESS_CACHE_MS; a revocation must not wait that out. */
export async function dropMcpAccess(repoId: string): Promise<void> {
  try {
    await env.MCP_HUB.get(env.MCP_HUB.idFromName(`repo:${repoId}`)).invalidateSubscribers(repoId);
  } catch (error) {
    console.error("MCP access invalidation failed", repoId, error);
  }
}

/** After a repository turns private: drop every follower who cannot read it any more. */
export async function purgeInaccessibleFollowers(repoId: string): Promise<void> {
  await purge([repoId], "t.repo_id = ?1", null, repoId);
  await dropMcpAccess(repoId);
}

/** After a collaborator is removed: drop that user's rows on the repository if they lost read access. */
export async function purgeFollowerIfRevoked(repoId: string, userId: string): Promise<void> {
  await purge([repoId], "t.repo_id = ?1", userId, repoId, userId);
  await revokeDeployTokensCreatedBy(userId, [repoId]);
  await dropMcpAccess(repoId);
}

/** After an org membership ends (or the org's default permission drops): across every repository the org owns. */
export async function purgeOrgFollowers(orgId: string, userId: string | null): Promise<void> {
  const repoFilter = "t.repo_id IN (SELECT id FROM repos WHERE owner_id = ?1)";
  const repos = await all<{ id: string }>("SELECT id FROM repos WHERE owner_id = ?", orgId);
  const repoIds = repos.map((repo) => repo.id);
  if (userId === null) await purge(repoIds, repoFilter, null, orgId);
  else {
    await purge(repoIds, repoFilter, userId, orgId, userId);
    await revokeDeployTokensCreatedBy(userId, repoIds);
  }
  await Promise.all(repos.map((repo) => dropMcpAccess(repo.id)));
}