feat(organizations): sync an organization that is already mirrored (#444)

Once an organization finished its first mirror the card said
"Successfully mirrored" and the menu only offered Mirror Options, Ignore
and Delete, so there was no way to refresh it by hand. The only options
left were waiting for the scheduler or syncing every repository one at a
time (#429).

Adds POST /api/job/sync-org, which takes one orgId and, in the
background, re-discovers the organization's repositories from its
source, mirrors the ones that were never mirrored and syncs the ones
that were. The organization sits at "mirroring" while the run is going
and settles on "mirrored" or "failed". Repositories another run already
owns, ignored ones and ones on their way out are left alone, and an
ignored or already running organization is refused.

The per-organization part of the scheduler's re-discovery moves to
src/lib/organization-discovery.ts so the route and the scheduler share
the identity rules instead of each having their own copy. The scheduler
does the same work as before.

In the UI the organization menu gains "Sync Organization" at the top for
mirrored and failed organizations, in both the mobile and the desktop
card layout.

src/components/organizations/Organization.tsx

@@ -21,6 +21,7 @@ import {
   SelectValue,
 } from "../ui/select";
 import type { MirrorOrgRequest, MirrorOrgResponse } from "@/types/mirror";
+import type { SyncOrgRequest, SyncOrgResponse } from "@/types/sync";
 import type { SourceProviderKind } from "@/lib/source-providers/kinds";
 import { useSSE } from "@/hooks/useSEE";
 import { useFilterParams } from "@/hooks/useFilterParams";
@@ -204,6 +205,62 @@ export function Organization() {
     }
   };
 
+  // Re-sync an organization that is already mirrored: the Mirror button only
+  // covers the first run, so this is the by-hand refresh (#429).
+  const handleSyncOrg = async ({ orgId }: { orgId: string }) => {
+    try {
+      if (!user || !user.id) {
+        return;
+      }
+
+      setLoadingOrgIds((prev) => new Set(prev).add(orgId));
+
+      const reqPayload: SyncOrgRequest = { orgId };
+
+      const response = await apiRequest<SyncOrgResponse>("/job/sync-org", {
+        method: "POST",
+        data: reqPayload,
+      });
+
+      if (response.success) {
+        const queued = response.queued;
+        const parts: string[] = [];
+        if (queued?.mirror) parts.push(`${queued.mirror} to mirror`);
+        if (queued?.sync) parts.push(`${queued.sync} to sync`);
+        toast.success(
+          parts.length > 0
+            ? `Sync started: ${parts.join(", ")}`
+            : "Sync started: checking for new repositories"
+        );
+
+        // Show the spinner right away; the card settles once the run reports back.
+        setOrganizations((prevOrgs) =>
+          prevOrgs.map((org) =>
+            org.id === orgId ? { ...org, status: "mirroring" } : org
+          )
+        );
+
+        // Refresh organization data to get updated repository breakdown
+        // Use a small delay to allow the backend to process the sync request
+        setTimeout(() => {
+          fetchOrganizations(true);
+        }, 1000);
+      } else {
+        toast.error(response.error || "Error starting sync job");
+      }
+    } catch (error) {
+      toast.error(
+        error instanceof Error ? error.message : "Error starting sync job"
+      );
+    } finally {
+      setLoadingOrgIds((prev) => {
+        const newSet = new Set(prev);
+        newSet.delete(orgId);
+        return newSet;
+      });
+    }
+  };
+
   const handleIgnoreOrg = async ({ orgId, ignore }: { orgId: string; ignore: boolean }) => {
     try {
       if (!user || !user.id) {
@@ -898,6 +955,7 @@ export function Organization() {
         setFilter={setFilter}
         loadingOrgIds={loadingOrgIds}
         onMirror={handleMirrorOrg}
+        onSync={handleSyncOrg}
         onIgnore={handleIgnoreOrg}
         onAddOrganization={() => setIsDialogOpen(true)}
         onDelete={handleRequestDeleteOrganization}

src/components/organizations/OrganizationsList.tsx

@@ -48,6 +48,12 @@ interface OrganizationListProps {
   filter: FilterParams;
   setFilter: (filter: FilterParams) => void;
   onMirror: ({ orgId }: { orgId: string }) => Promise<void>;
+  /**
+   * Re-sync an organization that has already been mirrored. The Mirror
+   * button only covers the first run, so without this an organization could
+   * never be refreshed by hand (#429).
+   */
+  onSync?: ({ orgId }: { orgId: string }) => Promise<void>;
   onIgnore?: ({ orgId, ignore }: { orgId: string; ignore: boolean }) => Promise<void>;
   loadingOrgIds: Set<string>;
   onAddOrganization?: () => void;
@@ -145,6 +151,7 @@ export function OrganizationList({
   filter,
   setFilter,
   onMirror,
+  onSync,
   onIgnore,
   loadingOrgIds,
   onAddOrganization,
@@ -726,6 +733,15 @@ export function OrganizationList({
                       </Button>
                     </DropdownMenuTrigger>
                     <DropdownMenuContent align="end">
+                      {onSync && (org.status === "mirrored" || org.status === "failed") && (
+                        <>
+                          <DropdownMenuItem onClick={() => org.id && onSync({ orgId: org.id })}>
+                            <RefreshCw className="h-4 w-4 mr-2" />
+                            Sync Organization
+                          </DropdownMenuItem>
+                          <DropdownMenuSeparator />
+                        </>
+                      )}
                       <DropdownMenuItem onClick={() => setOverridesTarget(org)}>
                         <SlidersHorizontal className="h-4 w-4 mr-2" />
                         Mirror Options
@@ -877,6 +893,15 @@ export function OrganizationList({
                       </Button>
                     </DropdownMenuTrigger>
                     <DropdownMenuContent align="end">
+                      {onSync && (org.status === "mirrored" || org.status === "failed") && (
+                        <>
+                          <DropdownMenuItem onClick={() => org.id && onSync({ orgId: org.id })}>
+                            <RefreshCw className="h-4 w-4 mr-2" />
+                            Sync Organization
+                          </DropdownMenuItem>
+                          <DropdownMenuSeparator />
+                        </>
+                      )}
                       <DropdownMenuItem onClick={() => setOverridesTarget(org)}>
                         <SlidersHorizontal className="h-4 w-4 mr-2" />
                         Mirror Options

src/lib/organization-discovery.ts

@@ -0,0 +1,171 @@
+/**
+ * Organization repository discovery.
+ *
+ * An organization pinned to a source can gain newly published upstream
+ * repositories at any time, and a public-only (tokenless) source never runs
+ * the personal listings that would otherwise surface them. The scheduler
+ * re-lists every included organization on each tick and the manual
+ * organization sync re-lists a single one, so both go through the function
+ * below and the identity rules (what already counts as tracked) live in one
+ * place.
+ *
+ * Repositories missing from the listing are deliberately NOT removed: a
+ * tokenless listing only shows public repositories, so cleanup stays
+ * membership-based (repository-cleanup-service).
+ */
+
+import { eq } from 'drizzle-orm';
+import { db, organizations, repositories } from '@/lib/db';
+import type { Config } from '@/lib/db/schema';
+import {
+  calcBatchSizeForInsert,
+  normalizeGitRepoToInsert,
+  repositoryIdentityKeys,
+  selectNewRepositoriesByIdentity,
+} from '@/lib/repo-utils';
+import { resolveOrganizationSkipForks } from '@/lib/utils/mirror-overrides';
+import type { SourceRecord } from '@/lib/sources';
+
+/** The organization fields re-discovery reads. */
+export interface DiscoverableOrganization {
+  id: string;
+  name: string;
+  sourceId?: string | null;
+  status?: string | null;
+  isIncluded?: boolean | null;
+  mirrorOverrides?: unknown;
+}
+
+export interface OrganizationRediscoveryResult {
+  /**
+   * False when nothing ran: the organization is ignored or excluded, or its
+   * pin resolves to no enabled source.
+   */
+  ran: boolean;
+  /** Repositories the source listed for the organization. */
+  listed: number;
+  /** New repository rows inserted by this pass. */
+  imported: number;
+  /** Listed repositories that another row already tracks on the same host. */
+  alreadyTracked: number;
+}
+
+const NOTHING_DISCOVERED: OrganizationRediscoveryResult = {
+  ran: false,
+  listed: 0,
+  imported: 0,
+  alreadyTracked: 0,
+};
+
+/**
+ * Re-list one organization on the source it is pinned to and insert the
+ * repositories that are not tracked yet (onConflictDoNothing keeps repeats
+ * idempotent). Unpinned organizations and pins whose source is gone keep the
+ * legacy DB-only behavior, so nothing changes for them.
+ */
+export async function rediscoverRepositoriesForOrganization({
+  config,
+  userId,
+  organization,
+  sources,
+  existingRepoKeys,
+  phase,
+  logPrefix = '[Scheduler]',
+}: {
+  config: Partial<Config> & { id: string };
+  userId: string;
+  organization: DiscoverableOrganization;
+  /** Enabled sources, when the caller already loaded them. */
+  sources?: SourceRecord[];
+  /**
+   * Identity keys of the repositories already tracked, when the caller
+   * already built them. Claimed keys are added to the set, so a caller that
+   * loops over organizations passes one set and never inserts the same
+   * upstream repository twice.
+   */
+  existingRepoKeys?: Set<string>;
+  /** Named in the log lines, e.g. "scheduled sync". */
+  phase: string;
+  logPrefix?: string;
+}): Promise<OrganizationRediscoveryResult> {
+  // An ignored organization is left alone entirely: rediscovering it would
+  // insert its new repositories as imported and auto-mirror could pick them
+  // up, which is what the user just asked to stop (#429).
+  if (organization.status === 'ignored' || organization.isIncluded === false) {
+    return NOTHING_DISCOVERED;
+  }
+
+  const { listSources, findSourceForOrganization } = await import('@/lib/sources');
+  const { createSourceProviderFromSource } = await import('@/lib/source-providers');
+
+  const sourceList = sources ?? (await listSources(userId)).filter(source => source.enabled);
+  const source = findSourceForOrganization(organization, sourceList);
+  if (!source) return NOTHING_DISCOVERED;
+
+  // Same identity rule as personal discovery: an organization pinned to a
+  // public-only source must not re-insert repositories the personal source
+  // already tracks on the same host (or the other way round).
+  const knownKeys = existingRepoKeys ?? repositoryIdentityKeys(
+    await db
+      .select({
+        normalizedFullName: repositories.normalizedFullName,
+        sourceProvider: repositories.sourceProvider,
+        sourceUrl: repositories.sourceUrl,
+      })
+      .from(repositories)
+      .where(eq(repositories.userId, userId))
+  );
+
+  const sourceProvider = createSourceProviderFromSource(source, { userId });
+  const orgRepos = await sourceProvider.listOrganizationRepositories(organization.name);
+
+  // The organization's fork policy (override -> global skipForks), the same
+  // resolution the organization mirror path applies.
+  const skipOrgForks = resolveOrganizationSkipForks({
+    orgOverrides: organization.mirrorOverrides,
+    config,
+  });
+  const mirrorableRepos = orgRepos.filter(
+    repo => repo.isDisabled !== true && !(skipOrgForks && repo.isForked)
+  );
+
+  const { fresh: repoRecords, alreadyTracked } = selectNewRepositoriesByIdentity(
+    mirrorableRepos.map(repo =>
+      normalizeGitRepoToInsert(
+        { ...repo, organization: repo.organization ?? organization.name },
+        { userId, configId: config.id, sourceId: source.id }
+      )
+    ),
+    knownKeys
+  );
+
+  if (repoRecords.length > 0) {
+    // Batch insert to avoid SQLite parameter limit
+    const sample = repoRecords[0];
+    const columnCount = Object.keys(sample ?? {}).length || 1;
+    const BATCH_SIZE = calcBatchSizeForInsert(columnCount);
+    for (let i = 0; i < repoRecords.length; i += BATCH_SIZE) {
+      const batch = repoRecords.slice(i, i + BATCH_SIZE);
+      await db
+        .insert(repositories)
+        .values(batch)
+        .onConflictDoNothing({ target: [repositories.userId, repositories.sourceId, repositories.normalizedFullName] });
+    }
+    console.log(`${logPrefix} Re-discovered ${repoRecords.length} new repositories for organization ${organization.name} on source ${source.name} for user ${userId} during ${phase}`);
+  }
+  if (alreadyTracked.length > 0) {
+    console.log(`${logPrefix} Organization ${organization.name}: ${alreadyTracked.length} repositories are already tracked for user ${userId} on the same host, not adding them again under source ${source.name}`);
+  }
+
+  await db
+    .update(organizations)
+    .set({ repositoryCount: orgRepos.length, updatedAt: new Date() })
+    .where(eq(organizations.id, organization.id));
+
+  return {
+    ran: true,
+    listed: orgRepos.length,
+    imported: repoRecords.length,
+    alreadyTracked: alreadyTracked.length,
+  };
+}

src/lib/organization-sync.ts

@@ -0,0 +1,479 @@
+/**
+ * Manual organization sync (issue #429).
+ *
+ * The organization card offers "Mirror Organization" only while an
+ * organization has never been mirrored. Once it reads "Successfully
+ * mirrored" there was no way to refresh it by hand: the user had to wait for
+ * the scheduler or sync every repository one at a time. This module is the
+ * by-hand version of what a scheduled run does for a single organization.
+ * It re-discovers the organization's repositories from its source, mirrors
+ * the ones that were never mirrored and syncs the ones that were, leaving
+ * rows another run already owns alone.
+ */
+
+import type { Octokit } from "@octokit/rest";
+import { and, eq } from "drizzle-orm";
+import { db, organizations, repositories } from "@/lib/db";
+import type { Config, Repository } from "@/lib/db/schema";
+import { usesPushEngine } from "@/lib/destination-connection";
+import { createGitHubClient, createPublicGitHubClient } from "@/lib/github";
+import { createMirrorJob } from "@/lib/helpers";
+import {
+  decryptSourceToken,
+  findSourceForOrganization,
+  findSourceForRepository,
+  listSources,
+  resolveGitHubApiBaseUrl,
+  type SourceRecord,
+} from "@/lib/sources";
+import { mirrorRepositoryToDestination, syncRepositoryOnDestination } from "@/lib/mirror-dispatch";
+import { rediscoverRepositoriesForOrganization, type DiscoverableOrganization } from "@/lib/organization-discovery";
+import { repoStatusEnum, repositoryVisibilityEnum } from "@/types/Repository";
+import { processWithResilience } from "@/lib/utils/concurrency";
+import { resolveOrganizationSkipForks } from "@/lib/utils/mirror-overrides";
+
+const LOG_PREFIX = "[SyncOrg]";
+
+/** Repository rows are processed a few at a time, as the other job routes do. */
+const MIRROR_CONCURRENCY_LIMIT = 3;
+const SYNC_CONCURRENCY_LIMIT = 5;
+
+/** Statuses a manual organization sync mirrors for the first time. */
+export const ORG_SYNC_MIRROR_STATUSES = ["imported"] as const;
+
+/**
+ * Statuses a manual organization sync refreshes on the destination. A failed
+ * row is synced rather than re-created: its mirror usually exists and only
+ * the metadata pass failed. Every other status is left alone, which is also
+ * how rows another run owns ("mirroring", "syncing") are skipped.
+ */
+export const ORG_SYNC_SYNC_STATUSES = ["mirrored", "synced", "failed"] as const;
+
+/** The organization fields a manual sync reads. */
+export interface SyncableOrganization extends DiscoverableOrganization {
+  id: string;
+  name: string;
+}
+
+type RepositoryRow = typeof repositories.$inferSelect;
+
+export interface OrganizationSyncPlan {
+  /** Never mirrored, so they get an initial mirror. */
+  toMirror: RepositoryRow[];
+  /** Already on the destination, so they get a refresh. */
+  toSync: RepositoryRow[];
+  /** In flight, ignored, being deleted or otherwise not ours to touch. */
+  skipped: RepositoryRow[];
+}
+
+export interface OrganizationSyncResult {
+  /** New repository rows the re-discovery pass imported. */
+  discovered: number;
+  /** Repositories mirrored for the first time by this run. */
+  mirrored: number;
+  /** Repositories refreshed on the destination by this run. */
+  synced: number;
+  /** Repositories another run had already claimed when the sync reached them. */
+  skippedInFlight: number;
+  /** Repositories that failed after their retries. */
+  failed: number;
+}
+
+/**
+ * Split an organization's repository rows into what a manual sync mirrors,
+ * refreshes and leaves alone.
+ *
+ * Forks are held back from the mirror set when the organization skips forks,
+ * the same rule mirrorGitHubOrgToGitea applies, so a run never creates a
+ * mirror the organization opted out of. A fork that was mirrored before the
+ * opt-out is still refreshed, so an existing mirror does not silently go
+ * stale.
+ */
+export function partitionOrganizationRepositories(
+  rows: RepositoryRow[],
+  { skipForks }: { skipForks: boolean }
+): OrganizationSyncPlan {
+  const plan: OrganizationSyncPlan = { toMirror: [], toSync: [], skipped: [] };
+
+  for (const row of rows) {
+    const status = row.status as (typeof ORG_SYNC_MIRROR_STATUSES)[number] | string;
+
+    if ((ORG_SYNC_MIRROR_STATUSES as readonly string[]).includes(status)) {
+      if (skipForks && row.isForked) {
+        plan.skipped.push(row);
+      } else {
+        plan.toMirror.push(row);
+      }
+      continue;
+    }
+
+    if ((ORG_SYNC_SYNC_STATUSES as readonly string[]).includes(status)) {
+      plan.toSync.push(row);
+      continue;
+    }
+
+    plan.skipped.push(row);
+  }
+
+  return plan;
+}
+
+/**
+ * The organization's repository rows. An organization pinned to one source
+ * only covers that source's repositories of the name, the same rule
+ * mirrorGitHubOrgToGitea applies; unpinned organizations follow every
+ * connected source.
+ */
+export async function selectOrganizationRepositoryRows({
+  userId,
+  organization,
+  sources,
+}: {
+  userId: string;
+  organization: SyncableOrganization;
+  sources: SourceRecord[];
+}): Promise<RepositoryRow[]> {
+  const pinnedSource = findSourceForOrganization(organization, sources);
+
+  const conditions = [
+    eq(repositories.userId, userId),
+    eq(repositories.organization, organization.name),
+  ];
+  if (pinnedSource) {
+    conditions.push(eq(repositories.sourceId, pinnedSource.id));
+  }
+
+  return db
+    .select()
+    .from(repositories)
+    .where(and(...conditions));
+}
+
+/** The row as the mirror and sync entry points expect it. */
+function toRepository(row: RepositoryRow): Repository {
+  return {
+    ...row,
+    status: repoStatusEnum.parse(row.status),
+    visibility: repositoryVisibilityEnum.parse(row.visibility),
+    organization: row.organization ?? undefined,
+    lastMirrored: row.lastMirrored ?? undefined,
+    errorMessage: row.errorMessage ?? undefined,
+    forkedFrom: row.forkedFrom ?? undefined,
+    mirroredLocation: row.mirroredLocation || "",
+  } as Repository;
+}
+
+/**
+ * A sync that found another run already owning the row returns a skip marker
+ * instead of throwing, so the counts can tell it apart from real work.
+ */
+function wasSkippedAsInFlight(result: unknown): boolean {
+  return (
+    !!result &&
+    typeof result === "object" &&
+    (result as { skipped?: unknown }).skipped === true
+  );
+}
+
+/**
+ * Only GitHub sources need an API client while mirroring (metadata); other
+ * hosts get code-only mirrors through the destination. One organization can
+ * hold repositories from several hosts, so each row is handed a client built
+ * from its own source.
+ */
+function createRepositoryOctokitResolver(
+  userId: string,
+  sources: SourceRecord[]
+): (repository: Repository) => Octokit | null {
+  return (repository) => {
+    const source = findSourceForRepository(repository, sources);
+    if (!source || source.provider !== "github") return null;
+
+    const token = decryptSourceToken(source.token);
+    // A tokenless GitHub source still gets an anonymous public client
+    // (60 req/hr) so public-repo metadata mirrors; an empty token must never
+    // reach createGitHubClient, which would send `auth: ""`.
+    return token
+      ? createGitHubClient(
+          token,
+          userId,
+          source.username || undefined,
+          resolveGitHubApiBaseUrl(source.url)
+        )
+      : createPublicGitHubClient(resolveGitHubApiBaseUrl(source.url));
+  };
+}
+
+/**
+ * Mirror one never-mirrored repository, the per-repository path retry-repo
+ * uses: a push destination creates the target itself, a pull destination
+ * needs the owning organization, which the org mirror entry point creates on
+ * demand.
+ */
+async function mirrorOneRepository({
+  config,
+  repository,
+  octokit,
+  sourceUsername,
+}: {
+  config: Partial<Config>;
+  repository: Repository;
+  octokit: Octokit | null;
+  sourceUsername?: string;
+}): Promise<void> {
+  if (usesPushEngine(config)) {
+    await mirrorRepositoryToDestination({ config, octokit, repository });
+    return;
+  }
+
+  const { getGiteaRepoOwnerAsync } = await import("@/lib/gitea");
+  const owner = await getGiteaRepoOwnerAsync({ config, repository, sourceUsername });
+
+  const mirrorStrategy =
+    config.githubConfig?.mirrorStrategy ||
+    (config.giteaConfig?.preserveOrgStructure ? "preserve" : "flat-user");
+  const shouldUseOrgMirror =
+    owner !== config.giteaConfig?.defaultOwner || mirrorStrategy === "single-org";
+
+  await mirrorRepositoryToDestination({
+    config,
+    octokit,
+    repository,
+    orgName: shouldUseOrgMirror ? owner : undefined,
+  });
+}
+
+/**
+ * Run a manual sync for one organization: re-discover, mirror what was never
+ * mirrored, refresh what was. The organization sits at "mirroring" for the
+ * duration and settles on "mirrored" or "failed".
+ */
+export async function runOrganizationSync({
+  config,
+  userId,
+  organization,
+}: {
+  config: Partial<Config> & { id: string };
+  userId: string;
+  organization: SyncableOrganization;
+}): Promise<OrganizationSyncResult> {
+  const startedAt = new Date();
+
+  await db
+    .update(organizations)
+    .set({ status: repoStatusEnum.parse("mirroring"), updatedAt: startedAt })
+    .where(eq(organizations.id, organization.id));
+
+  await createMirrorJob({
+    userId,
+    organizationId: organization.id,
+    organizationName: organization.name,
+    message: `Started sync for organization: ${organization.name}`,
+    details: `Organization ${organization.name} is being re-discovered, mirrored and synced.`,
+    status: repoStatusEnum.parse("mirroring"),
+    jobType: "sync",
+  });
+
+  const result: OrganizationSyncResult = {
+    discovered: 0,
+    mirrored: 0,
+    synced: 0,
+    skippedInFlight: 0,
+    failed: 0,
+  };
+  const notes: string[] = [];
+
+  try {
+    // 1. New upstream repositories. A source that cannot be listed must not
+    // stop the refresh of everything already mirrored, so the failure is
+    // reported in the job details instead of ending the run.
+    try {
+      const discovery = await rediscoverRepositoriesForOrganization({
+        config,
+        userId,
+        organization,
+        phase: "manual organization sync",
+        logPrefix: LOG_PREFIX,
+      });
+      result.discovered = discovery.imported;
+    } catch (error) {
+      const message = error instanceof Error ? error.message : String(error);
+      console.error(
+        `${LOG_PREFIX} Failed to re-discover repositories for organization ${organization.name}: ${message}`
+      );
+      notes.push(`Re-discovery failed: ${message}`);
+    }
+
+    // 2. Plan over the rows as they stand now, the newly imported included.
+    const sources = await listSources(userId);
+    const rows = await selectOrganizationRepositoryRows({ userId, organization, sources });
+    const skipForks = resolveOrganizationSkipForks({
+      orgOverrides: organization.mirrorOverrides,
+      config,
+    });
+    const { toMirror, toSync, skipped } = partitionOrganizationRepositories(rows, { skipForks });
+
+    console.log(
+      `${LOG_PREFIX} Organization ${organization.name}: mirroring ${toMirror.length}, syncing ${toSync.length}, skipping ${skipped.length} repositories`
+    );
+
+    const resolveRepositoryOctokit = createRepositoryOctokitResolver(userId, sources);
+
+    // 3. Initial mirrors.
+    if (toMirror.length > 0) {
+      const mirrored = await processWithResilience(
+        toMirror,
+        async (row) => {
+          const repository = toRepository(row);
+          const source = findSourceForRepository(repository, sources);
+          console.log(`${LOG_PREFIX} Starting mirror for repository: ${row.name}`);
+          await mirrorOneRepository({
+            config,
+            repository,
+            octokit: resolveRepositoryOctokit(repository),
+            sourceUsername: source?.username,
+          });
+          return row;
+        },
+        {
+          userId,
+          jobType: "mirror",
+          getItemId: (row) => row.id,
+          getItemName: (row) => row.name,
+          concurrencyLimit: MIRROR_CONCURRENCY_LIMIT,
+          maxRetries: 2,
+          retryDelay: 2000,
+          checkpointInterval: 1,
+          onProgress: (completed, total) => {
+            const percentComplete = Math.round((completed / total) * 100);
+            console.log(
+              `${LOG_PREFIX} Mirror progress for ${organization.name}: ${percentComplete}% (${completed}/${total})`
+            );
+          },
+          onRetry: (row, error, attempt) => {
+            console.log(
+              `${LOG_PREFIX} Retrying mirror for repository ${row.name} (attempt ${attempt}): ${error.message}`
+            );
+          },
+        }
+      );
+
+      result.mirrored = mirrored.length;
+      result.failed += toMirror.length - mirrored.length;
+    }
+
+    // 4. Refresh of the existing mirrors.
+    if (toSync.length > 0) {
+      const synced = await processWithResilience(
+        toSync,
+        async (row) => {
+          const repository = toRepository(row);
+          console.log(`${LOG_PREFIX} Starting sync for repository: ${row.name}`);
+          const syncResult = await syncRepositoryOnDestination({ config, repository });
+          return { row, skipped: wasSkippedAsInFlight(syncResult) };
+        },
+        {
+          userId,
+          jobType: "sync",
+          getItemId: (row) => row.id,
+          getItemName: (row) => row.name,
+          concurrencyLimit: SYNC_CONCURRENCY_LIMIT,
+          maxRetries: 2,
+          retryDelay: 2000,
+          checkpointInterval: 1,
+          onProgress: (completed, total) => {
+            const percentComplete = Math.round((completed / total) * 100);
+            console.log(
+              `${LOG_PREFIX} Sync progress for ${organization.name}: ${percentComplete}% (${completed}/${total})`
+            );
+          },
+          onRetry: (row, error, attempt) => {
+            console.log(
+              `${LOG_PREFIX} Retrying sync for repository ${row.name} (attempt ${attempt}): ${error.message}`
+            );
+          },
+        }
+      );
+
+      result.skippedInFlight = synced.filter((entry) => entry.skipped).length;
+      result.synced = synced.length - result.skippedInFlight;
+      result.failed += toSync.length - synced.length;
+    }
+
+    const summary =
+      `${result.mirrored} mirrored, ${result.synced} synced, ${result.discovered} newly discovered, ` +
+      `${result.skippedInFlight} already in progress, ${result.failed} failed.`;
+    const details = [`Organization ${organization.name}: ${summary}`, ...notes].join(" ");
+
+    if (result.failed > 0) {
+      await db
+        .update(organizations)
+        .set({
+          status: repoStatusEnum.parse("failed"),
+          updatedAt: new Date(),
+          errorMessage: `${result.failed} ${result.failed === 1 ? "repository" : "repositories"} failed to sync.`,
+        })
+        .where(eq(organizations.id, organization.id));
+
+      await createMirrorJob({
+        userId,
+        organizationId: organization.id,
+        organizationName: organization.name,
+        message: `Failed to sync organization: ${organization.name}`,
+        details,
+        status: repoStatusEnum.parse("failed"),
+        jobType: "sync",
+      });
+
+      return result;
+    }
+
+    await db
+      .update(organizations)
+      .set({
+        status: repoStatusEnum.parse("mirrored"),
+        updatedAt: new Date(),
+        lastMirrored: new Date(),
+        errorMessage: null,
+      })
+      .where(eq(organizations.id, organization.id));
+
+    await createMirrorJob({
+      userId,
+      organizationId: organization.id,
+      organizationName: organization.name,
+      message: `Successfully synced organization: ${organization.name}`,
+      details,
+      status: repoStatusEnum.parse("mirrored"),
+      jobType: "sync",
+    });
+
+    return result;
+  } catch (error) {
+    const message = error instanceof Error ? error.message : "Unknown error";
+    console.error(
+      `${LOG_PREFIX} Error while syncing organization ${organization.name}: ${message}`
+    );
+
+    await db
+      .update(organizations)
+      .set({
+        status: repoStatusEnum.parse("failed"),
+        updatedAt: new Date(),
+        errorMessage: message,
+      })
+      .where(eq(organizations.id, organization.id));
+
+    await createMirrorJob({
+      userId,
+      organizationId: organization.id,
+      organizationName: organization.name,
+      message: `Failed to sync organization: ${organization.name}`,
+      details: `Organization ${organization.name} failed to sync. Error: ${message}`,
+      status: repoStatusEnum.parse("failed"),
+      jobType: "sync",
+    });
+
+    return result;
+  }
+}

src/lib/scheduler-service.ts

@@ -22,8 +22,9 @@ import {
   dedupeRepositoriesByIdentity,
 } from '@/lib/repo-utils';
 import { isMirrorableGitHubRepo } from '@/lib/repo-eligibility';
-import { loadOrganizationForkPolicies, orgForkSkipDecision, resolveOrganizationSkipForks } from '@/lib/utils/mirror-overrides';
+import { loadOrganizationForkPolicies, orgForkSkipDecision } from '@/lib/utils/mirror-overrides';
 import { createMirrorJob } from '@/lib/helpers';
+import { rediscoverRepositoriesForOrganization } from '@/lib/organization-discovery';
 import { getNextScheduledRun, isCronExpression, normalizeTimezone } from '@/lib/utils/schedule-utils';
 import { resetStuckMirrorStatuses } from '@/lib/stuck-status-recovery';
 import { isRateLimitError, rateLimitGateKey, rateLimitedUntil } from '@/lib/rate-limit-gate';
@@ -295,26 +296,18 @@ async function importRepositoriesFromSources(
 }
 
 /**
- * Re-discover organization repositories: an organization pinned to a source
+ * Re-discover organization repositories: each tick re-lists every included
- * can gain newly published upstream repositories at any time, and a
- * public-only (tokenless) source never runs the personal listings that
- * would otherwise surface them. Each tick re-lists every included
  * organization whose pin resolves to an existing enabled source and inserts
- * the repositories it does not have yet (onConflictDoNothing keeps repeats
+ * the repositories it does not have yet. The per-organization work lives in
- * idempotent). Unpinned organizations and pins whose source is gone keep
+ * rediscoverRepositoriesForOrganization, which the manual organization sync
- * the legacy DB-only behavior — no behavior change for existing users.
+ * route shares.
- *
- * Repositories missing from the listing are deliberately NOT removed:
- * a tokenless listing only shows public repositories, so cleanup stays
- * membership-based (repository-cleanup-service).
  */
 async function rediscoverOrganizationRepositories(
   config: any,
   userId: string,
   phase: 'scheduled sync' | 'auto-start'
 ): Promise<void> {
-  const { listSources, findSourceForOrganization } = await import('@/lib/sources');
+  const { listSources } = await import('@/lib/sources');
-  const { createSourceProviderFromSource } = await import('@/lib/source-providers');
   const sources = (await listSources(userId)).filter(source => source.enabled);
 
   // An ignored organization is left alone entirely: rediscovering it would
@@ -333,7 +326,9 @@ async function rediscoverOrganizationRepositories(
 
   // Same identity rule as personal discovery: an organization pinned to a
   // public-only source must not re-insert repositories the personal source
-  // already tracks on the same host (or the other way round).
+  // already tracks on the same host (or the other way round). One set is
+  // shared by every organization in the pass, so it also keeps two
+  // organizations from inserting the same repository twice.
   const existingRepos = await db
     .select({
       normalizedFullName: repositories.normalizedFullName,
@@ -346,51 +341,14 @@ async function rediscoverOrganizationRepositories(
 
   for (const org of includedOrgs) {
     try {
-      const source = findSourceForOrganization(org, sources);
+      await rediscoverRepositoriesForOrganization({
-      if (!source) continue;
+        config,
-
+        userId,
-      const sourceProvider = createSourceProviderFromSource(source, { userId });
+        organization: org,
-      const orgRepos = await sourceProvider.listOrganizationRepositories(org.name);
+        sources,
-
+        existingRepoKeys,
-      // The organization's fork policy (override -> global skipForks), the
+        phase,
-      // same resolution the organization mirror path applies.
+      });
-      const skipOrgForks = resolveOrganizationSkipForks({ orgOverrides: org.mirrorOverrides, config });
-      const mirrorableRepos = orgRepos.filter(
-        repo => repo.isDisabled !== true && !(skipOrgForks && repo.isForked)
-      );
-
-      const { fresh: repoRecords, alreadyTracked } = selectNewRepositoriesByIdentity(
-        mirrorableRepos.map(repo =>
-          normalizeGitRepoToInsert(
-            { ...repo, organization: repo.organization ?? org.name },
-            { userId, configId: config.id, sourceId: source.id }
-          )
-        ),
-        existingRepoKeys
-      );
-
-      if (repoRecords.length > 0) {
-        // Batch insert to avoid SQLite parameter limit
-        const sample = repoRecords[0];
-        const columnCount = Object.keys(sample ?? {}).length || 1;
-        const BATCH_SIZE = calcBatchSizeForInsert(columnCount);
-        for (let i = 0; i < repoRecords.length; i += BATCH_SIZE) {
-          const batch = repoRecords.slice(i, i + BATCH_SIZE);
-          await db
-            .insert(repositories)
-            .values(batch)
-            .onConflictDoNothing({ target: [repositories.userId, repositories.sourceId, repositories.normalizedFullName] });
-        }
-        console.log(`[Scheduler] Re-discovered ${repoRecords.length} new repositories for organization ${org.name} on source ${source.name} for user ${userId} during ${phase}`);
-      }
-      if (alreadyTracked.length > 0) {
-        console.log(`[Scheduler] Organization ${org.name}: ${alreadyTracked.length} repositories are already tracked for user ${userId} on the same host, not adding them again under source ${source.name}`);
-      }
-
-      await db
-        .update(organizations)
-        .set({ repositoryCount: orgRepos.length, updatedAt: new Date() })
-        .where(eq(organizations.id, org.id));
     } catch (orgError) {
       console.error(`[Scheduler] Failed to re-discover repositories for organization ${org.name} for user ${userId} during ${phase}:`, orgError);
     }

src/pages/api/job/sync-org.test.ts

@@ -0,0 +1,468 @@
+/**
+ * Route tests for POST /api/job/sync-org: the manual re-sync of one
+ * organization (#429). The route re-discovers the organization's
+ * repositories, mirrors the ones that were never mirrored and refreshes the
+ * ones that were, while the organization sits at "mirroring".
+ *
+ * Runs against the real schema in an in-memory SQLite database. The route
+ * imports @/lib/db, which is replaced with that database through
+ * mock.module; bun's mock.module is process-wide and leaks into other test
+ * files, so this file registers nothing in the shared process and re-runs
+ * itself in an isolated child (same harness as
+ * ../organizations/[id]/status.test.ts).
+ */
+import { describe, test, expect, mock, beforeEach } from "bun:test";
+import { Database } from "bun:sqlite";
+import { drizzle } from "drizzle-orm/bun-sqlite";
+import { readFileSync } from "node:fs";
+import { join } from "node:path";
+import * as schema from "@/lib/db/schema";
+
+const CHILD_FLAG = "GM_SYNC_ORG_ROUTE_ISOLATED";
+const isChild = !!process.env[CHILD_FLAG];
+
+if (!isChild) {
+  test("organization sync route - isolated child suite", () => {
+    const res = Bun.spawnSync({
+      cmd: [process.execPath, "test", import.meta.path],
+      env: { ...process.env, [CHILD_FLAG]: "1" },
+      stdout: "pipe",
+      stderr: "pipe",
+    });
+    if (res.exitCode !== 0) {
+      console.error(res.stdout.toString());
+      console.error(res.stderr.toString());
+    }
+    expect(res.exitCode).toBe(0);
+  }, 120_000);
+}
+
+const sqlite = new Database(":memory:");
+const db = drizzle({ client: sqlite });
+
+/**
+ * Apply every migration in drizzle/ in journal order. The shared test setup
+ * stubs drizzle's migrator, so the SQL is run here the way the migrator
+ * would: one file per journal entry, statements split on its breakpoint.
+ */
+function applyMigrations(): void {
+  const folder = join(process.cwd(), "drizzle");
+  const journal = JSON.parse(readFileSync(join(folder, "meta", "_journal.json"), "utf8")) as {
+    entries: { tag: string }[];
+  };
+  for (const entry of journal.entries) {
+    const sql = readFileSync(join(folder, `${entry.tag}.sql`), "utf8");
+    for (const statement of sql.split("--> statement-breakpoint")) {
+      if (statement.trim()) sqlite.run(statement);
+    }
+  }
+}
+
+/** Repositories the mirror entry point was called for, in call order. */
+const mirrorCalls: string[] = [];
+/** Repositories the sync entry point was called for, in call order. */
+const syncCalls: string[] = [];
+/** Repository names whose sync must throw, and the ones that report a skip. */
+const syncFailures = new Set<string>();
+const syncSkips = new Set<string>();
+/**
+ * The organization-level job records the run writes. The shared test setup
+ * stubs @/lib/helpers, so the mirror_jobs rows (and the realtime events
+ * createMirrorJob publishes from them) cannot be read back from the
+ * database; this spy stands in for both.
+ */
+const jobCalls: {
+  status: string;
+  jobType?: string;
+  organizationId?: string;
+  message: string;
+  details?: string;
+}[] = [];
+/** Rows the (mocked) re-discovery pass inserts before the run plans its work. */
+let discoveryInserts: string[] = [];
+let discoveryThrows = false;
+
+const sourceRows = [
+  {
+    id: "source-github",
+    userId: "user-1",
+    name: "GitHub",
+    provider: "github",
+    url: "https://github.com",
+    username: "octocat",
+    token: "source-token",
+    enabled: true,
+    createdAt: new Date(0),
+    updatedAt: new Date(0),
+  },
+];
+
+if (isChild) {
+  applyMigrations();
+  // The seeds below only cover the tables the route touches.
+  sqlite.run("PRAGMA foreign_keys = OFF");
+
+  mock.module("@/lib/auth-guards", () => ({
+    requireAuthenticatedUserId: mock(async () => ({ userId: "user-1" })),
+  }));
+
+  mock.module("@/lib/db", () => ({ ...schema, db }));
+
+  mock.module("@/lib/helpers", () => ({
+    createMirrorJob: mock(async (job: (typeof jobCalls)[number]) => {
+      jobCalls.push(job);
+      return "job-id";
+    }),
+    createEvent: mock(async () => {}),
+    updateMirrorJobProgress: mock(async () => {}),
+    touchMirrorJobCheckpoint: mock(async () => {}),
+  }));
+
+  mock.module("@/lib/sources", () => ({
+    listSources: mock(async () => sourceRows),
+    findSourceForOrganization: (org: { sourceId?: string | null }) =>
+      org.sourceId ? sourceRows.find((source) => source.id === org.sourceId) ?? null : null,
+    findSourceForRepository: () => sourceRows[0],
+    decryptSourceToken: (token: string | null | undefined) => token ?? "",
+    resolveGitHubApiBaseUrl: () => undefined,
+    ensureSourcesFromConfig: mock(async () => {}),
+  }));
+
+  mock.module("@/lib/github", () => ({
+    createGitHubClient: mock(() => ({ authenticated: true })),
+    createPublicGitHubClient: mock(() => ({ anonymous: true })),
+  }));
+
+  mock.module("@/lib/gitea", () => ({
+    getGiteaRepoOwnerAsync: mock(async () => "acme"),
+  }));
+
+  mock.module("@/lib/mirror-dispatch", () => ({
+    mirrorRepositoryToDestination: mock(async ({ repository }: { repository: { name: string } }) => {
+      mirrorCalls.push(repository.name);
+      if (syncFailures.has(repository.name)) {
+        throw new Error(`mirror of ${repository.name} failed`);
+      }
+      return {};
+    }),
+    syncRepositoryOnDestination: mock(async ({ repository }: { repository: { name: string } }) => {
+      syncCalls.push(repository.name);
+      if (syncFailures.has(repository.name)) {
+        throw new Error(`sync of ${repository.name} failed`);
+      }
+      if (syncSkips.has(repository.name)) {
+        return { skipped: true, reason: "already-in-progress" };
+      }
+      return {};
+    }),
+  }));
+
+  mock.module("@/lib/organization-discovery", () => ({
+    rediscoverRepositoriesForOrganization: mock(async () => {
+      if (discoveryThrows) throw new Error("source unreachable");
+      for (const name of discoveryInserts) {
+        seedRepo({ id: name, organization: "acme", status: "imported" });
+      }
+      return {
+        ran: true,
+        listed: discoveryInserts.length,
+        imported: discoveryInserts.length,
+        alreadyTracked: 0,
+      };
+    }),
+  }));
+
+  // The real helper retries with exponential backoff and drops the items
+  // that never succeed from its results; this runs the same contract without
+  // the waiting.
+  mock.module("@/lib/utils/concurrency", () => ({
+    processWithResilience: mock(async (items: unknown[], worker: (item: unknown) => Promise<unknown>) => {
+      const results: unknown[] = [];
+      for (const item of items) {
+        try {
+          results.push(await worker(item));
+        } catch {
+          // Dropped, exactly as Promise.allSettled does in the real helper.
+        }
+      }
+      return results;
+    }),
+  }));
+}
+
+const { POST } = isChild ? await import("./sync-org") : { POST: undefined as any };
+
+function post(body: Record<string, unknown>) {
+  return POST({
+    request: new Request("http://localhost/api/job/sync-org", {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify(body),
+    }),
+    locals: { session: { userId: "user-1" } },
+  } as any);
+}
+
+type RepoSeed = {
+  id: string;
+  userId?: string;
+  organization: string | null;
+  status: string;
+  isForked?: boolean;
+  sourceId?: string | null;
+};
+
+function seedRepo({
+  id,
+  userId = "user-1",
+  organization,
+  status,
+  isForked = false,
+  sourceId = "source-github",
+}: RepoSeed) {
+  const owner = organization ?? "octocat";
+  sqlite.run(
+    `INSERT INTO repositories (id, user_id, config_id, name, full_name, normalized_full_name, url, clone_url, owner, organization, default_branch, status, is_fork, source_id)
+     VALUES (?, ?, 'config-1', ?, ?, ?, ?, ?, ?, ?, 'main', ?, ?, ?)`,
+    [
+      id,
+      userId,
+      id,
+      `${owner}/${id}`,
+      `${owner}/${id}`.toLowerCase(),
+      `https://github.com/${owner}/${id}`,
+      `https://github.com/${owner}/${id}.git`,
+      owner,
+      organization,
+      status,
+      isForked ? 1 : 0,
+      sourceId,
+    ]
+  );
+}
+
+function seedConfig(): void {
+  sqlite.run(
+    `INSERT INTO configs (id, user_id, name, github_config, gitea_config, schedule_config, cleanup_config)
+     VALUES ('config-1', 'user-1', 'default', ?, ?, '{}', '{}')`,
+    [
+      JSON.stringify({ owner: "octocat", mirrorStrategy: "preserve", skipForks: false }),
+      JSON.stringify({ url: "https://gitea.example.com", token: "gitea-token", defaultOwner: "octocat" }),
+    ]
+  );
+}
+
+function seedOrg({
+  id,
+  userId = "user-1",
+  name,
+  status,
+  mirrorOverrides = null,
+}: {
+  id: string;
+  userId?: string;
+  name: string;
+  status: string;
+  mirrorOverrides?: string | null;
+}) {
+  sqlite.run(
+    `INSERT INTO organizations (id, user_id, config_id, name, normalized_name, avatar_url, status, source_id, mirror_overrides)
+     VALUES (?, ?, 'config-1', ?, ?, '', ?, 'source-github', ?)`,
+    [id, userId, name, name.toLowerCase(), status, mirrorOverrides]
+  );
+}
+
+function orgRow(id: string): { status: string; last_mirrored: number | null; error_message: string | null } {
+  return sqlite
+    .query("SELECT status, last_mirrored, error_message FROM organizations WHERE id = ?")
+    .get(id) as any;
+}
+
+/** Wait until the background run has settled the organization's status. */
+async function waitForSettled(id: string, timeoutMs = 5000): Promise<string> {
+  const deadline = Date.now() + timeoutMs;
+  for (;;) {
+    const status = orgRow(id)?.status;
+    if (status && status !== "mirroring") return status;
+    if (Date.now() > deadline) {
+      throw new Error(`timed out waiting for organization ${id} to settle (status ${status})`);
+    }
+    await new Promise((resolve) => setTimeout(resolve, 10));
+  }
+}
+
+beforeEach(() => {
+  if (!isChild) return;
+  sqlite.run("DELETE FROM repositories");
+  sqlite.run("DELETE FROM organizations");
+  sqlite.run("DELETE FROM configs");
+  mirrorCalls.length = 0;
+  syncCalls.length = 0;
+  jobCalls.length = 0;
+  syncFailures.clear();
+  syncSkips.clear();
+  discoveryInserts = [];
+  discoveryThrows = false;
+
+  seedConfig();
+  seedOrg({ id: "org-acme", name: "acme", status: "mirrored" });
+  seedOrg({ id: "org-theirs", userId: "user-2", name: "theirs", status: "mirrored" });
+
+  seedRepo({ id: "acme-imported", organization: "acme", status: "imported" });
+  seedRepo({ id: "acme-mirrored", organization: "acme", status: "mirrored" });
+  seedRepo({ id: "acme-synced", organization: "acme", status: "synced" });
+  seedRepo({ id: "acme-failed", organization: "acme", status: "failed" });
+  // Owned by another run, on the way out, or opted out: left alone.
+  seedRepo({ id: "acme-mirroring", organization: "acme", status: "mirroring" });
+  seedRepo({ id: "acme-syncing", organization: "acme", status: "syncing" });
+  seedRepo({ id: "acme-deleting", organization: "acme", status: "deleting" });
+  seedRepo({ id: "acme-ignored", organization: "acme", status: "ignored" });
+  // Another organization and a personal repository: not this organization's.
+  seedRepo({ id: "beta-mirrored", organization: "beta", status: "mirrored" });
+  seedRepo({ id: "personal-mirrored", organization: null, status: "mirrored" });
+});
+
+describe.skipIf(!isChild)("POST /api/job/sync-org", () => {
+  test("rejects a request without an orgId", async () => {
+    const response = await post({});
+    expect(response.status).toBe(400);
+    const data = await response.json();
+    expect(data.success).toBe(false);
+    expect(data.error).toBe("orgId is required.");
+  });
+
+  test("answers 404 for another user's organization", async () => {
+    const response = await post({ orgId: "org-theirs" });
+    expect(response.status).toBe(404);
+    expect(orgRow("org-theirs").status).toBe("mirrored");
+    expect(mirrorCalls).toEqual([]);
+    expect(syncCalls).toEqual([]);
+  });
+
+  test("refuses an ignored organization and touches nothing", async () => {
+    sqlite.run("UPDATE organizations SET status = 'ignored' WHERE id = 'org-acme'");
+
+    const response = await post({ orgId: "org-acme" });
+    expect(response.status).toBe(409);
+    const data = await response.json();
+    expect(data.error).toContain("ignored");
+    expect(orgRow("org-acme").status).toBe("ignored");
+    expect(mirrorCalls).toEqual([]);
+    expect(syncCalls).toEqual([]);
+  });
+
+  test("refuses an organization that is already being processed", async () => {
+    sqlite.run("UPDATE organizations SET status = 'mirroring' WHERE id = 'org-acme'");
+
+    const response = await post({ orgId: "org-acme" });
+    expect(response.status).toBe(409);
+    const data = await response.json();
+    expect(data.error).toContain("already being processed");
+    expect(mirrorCalls).toEqual([]);
+    expect(syncCalls).toEqual([]);
+  });
+
+  test("answers with the queued counts and reports the organization as mirroring", async () => {
+    const response = await post({ orgId: "org-acme" });
+    expect(response.status).toBe(200);
+
+    const data = await response.json();
+    expect(data.success).toBe(true);
+    expect(data.message).toBe("Sync job started.");
+    expect(data.organization.id).toBe("org-acme");
+    expect(data.organization.status).toBe("mirroring");
+    // One imported, three already on the destination, four left alone.
+    expect(data.queued).toEqual({ mirror: 1, sync: 3, skipped: 4 });
+
+    await waitForSettled("org-acme");
+  });
+
+  test("mirrors what was never mirrored, syncs the rest and skips the others", async () => {
+    await post({ orgId: "org-acme" });
+    expect(await waitForSettled("org-acme")).toBe("mirrored");
+
+    expect(mirrorCalls).toEqual(["acme-imported"]);
+    expect(syncCalls.sort()).toEqual(["acme-failed", "acme-mirrored", "acme-synced"]);
+
+    const settled = orgRow("org-acme");
+    expect(settled.last_mirrored).not.toBeNull();
+    expect(settled.error_message).toBeNull();
+  });
+
+  test("picks up the repositories the re-discovery pass imported", async () => {
+    discoveryInserts = ["acme-brand-new"];
+
+    const response = await post({ orgId: "org-acme" });
+    const data = await response.json();
+    // The answer is a snapshot of the rows as they stood: the new row is
+    // discovered by the run itself and mirrored on top of the promise.
+    expect(data.queued.mirror).toBe(1);
+
+    expect(await waitForSettled("org-acme")).toBe("mirrored");
+    expect(mirrorCalls.sort()).toEqual(["acme-brand-new", "acme-imported"]);
+  });
+
+  test("keeps going when the re-discovery pass fails", async () => {
+    discoveryThrows = true;
+
+    await post({ orgId: "org-acme" });
+    expect(await waitForSettled("org-acme")).toBe("mirrored");
+
+    expect(mirrorCalls).toEqual(["acme-imported"]);
+    expect(syncCalls.length).toBe(3);
+
+    const done = jobCalls.find((job) => job.status === "mirrored");
+    expect(done?.details).toContain("Re-discovery failed");
+  });
+
+  test("holds forks back from the mirror set when the organization skips forks", async () => {
+    sqlite.run("DELETE FROM repositories WHERE id = 'acme-imported'");
+    seedRepo({ id: "acme-fork", organization: "acme", status: "imported", isForked: true });
+    sqlite.run(
+      "UPDATE organizations SET mirror_overrides = ? WHERE id = 'org-acme'",
+      [JSON.stringify({ skipForks: true })]
+    );
+
+    const response = await post({ orgId: "org-acme" });
+    const data = await response.json();
+    expect(data.queued).toEqual({ mirror: 0, sync: 3, skipped: 5 });
+
+    expect(await waitForSettled("org-acme")).toBe("mirrored");
+    expect(mirrorCalls).toEqual([]);
+  });
+
+  test("a repository another run owns counts as in progress, not as a failure", async () => {
+    syncSkips.add("acme-mirrored");
+
+    await post({ orgId: "org-acme" });
+    expect(await waitForSettled("org-acme")).toBe("mirrored");
+
+    const done = jobCalls.find((job) => job.status === "mirrored");
+    expect(done?.details).toContain("1 already in progress");
+    expect(done?.details).toContain("0 failed");
+  });
+
+  test("marks the organization failed when a repository does not make it", async () => {
+    syncFailures.add("acme-failed");
+
+    await post({ orgId: "org-acme" });
+    expect(await waitForSettled("org-acme")).toBe("failed");
+
+    const settled = orgRow("org-acme");
+    expect(settled.error_message).toContain("1 repository failed");
+
+    const failedJob = jobCalls.find((job) => job.status === "failed");
+    expect(failedJob?.message).toContain("Failed to sync organization");
+  });
+
+  test("records the start and the end of the run as organization jobs", async () => {
+    await post({ orgId: "org-acme" });
+    expect(await waitForSettled("org-acme")).toBe("mirrored");
+
+    // Both carry the organization, so the realtime event createMirrorJob
+    // publishes reaches the card and the UI follows the run.
+    expect(jobCalls.map((job) => job.status)).toEqual(["mirroring", "mirrored"]);
+    expect(jobCalls.every((job) => job.jobType === "sync")).toBe(true);
+    expect(jobCalls.every((job) => job.organizationId === "org-acme")).toBe(true);
+  });
+});

src/pages/api/job/sync-org.ts

@@ -0,0 +1,147 @@
+import type { APIRoute } from "astro";
+import { db, configs, organizations } from "@/lib/db";
+import { and, eq, notInArray, sql } from "drizzle-orm";
+import { listSources } from "@/lib/sources";
+import {
+  partitionOrganizationRepositories,
+  runOrganizationSync,
+  selectOrganizationRepositoryRows,
+} from "@/lib/organization-sync";
+import { resolveOrganizationSkipForks } from "@/lib/utils/mirror-overrides";
+import { repoStatusEnum } from "@/types/Repository";
+import { type MembershipRole } from "@/types/organizations";
+import type { SyncOrgRequest, SyncOrgResponse } from "@/types/sync";
+import { createSecureErrorResponse } from "@/lib/utils";
+import { requireAuthenticatedUserId } from "@/lib/auth-guards";
+
+function refusal(message: string, status: number) {
+  // `message` is echoed as well as `error`: the browser client turns a
+  // non-2xx body into "HTTP <status>: <message>", so the toast reads as a
+  // sentence instead of a dump of the JSON body.
+  return new Response(JSON.stringify({ success: false, error: message, message }), {
+    status,
+    headers: { "Content-Type": "application/json" },
+  });
+}
+
+export const POST: APIRoute = async ({ request, locals }) => {
+  try {
+    const authResult = await requireAuthenticatedUserId({ request, locals });
+    if ("response" in authResult) return authResult.response;
+    const userId = authResult.userId;
+
+    const body: SyncOrgRequest = await request.json();
+    const { orgId } = body;
+
+    if (!orgId || typeof orgId !== "string") {
+      return refusal("orgId is required.", 400);
+    }
+
+    // Fetch config: prefer active and most-recently-updated to avoid picking
+    // a stale inactive stub when multiple rows exist (see issue #271).
+    const configResult = await db
+      .select()
+      .from(configs)
+      .where(eq(configs.userId, userId))
+      .orderBy(sql`${configs.isActive} DESC`, sql`${configs.updatedAt} DESC`)
+      .limit(1);
+
+    const config = configResult[0];
+
+    // Syncing only touches the destination, and a public-only source has no
+    // token at all, so only the destination token is required here.
+    if (!config || !config.giteaConfig?.token) {
+      return refusal("Config missing for the user or destination token.", 400);
+    }
+
+    const [organization] = await db
+      .select()
+      .from(organizations)
+      .where(and(eq(organizations.userId, userId), eq(organizations.id, orgId)))
+      .limit(1);
+
+    if (!organization) {
+      return refusal("No organization found for the given ID.", 404);
+    }
+
+    if (organization.status === "ignored") {
+      return refusal(
+        `Organization ${organization.name} is ignored. Include it again before syncing.`,
+        409
+      );
+    }
+
+    if (organization.status === "mirroring") {
+      return refusal(`Organization ${organization.name} is already being processed.`, 409);
+    }
+
+    // The counts are a snapshot of the rows as they stand now. Repositories
+    // the re-discovery pass imports are mirrored by the same run, so the run
+    // can do a little more than the response promises, never less.
+    const sources = await listSources(userId);
+    const rows = await selectOrganizationRepositoryRows({ userId, organization, sources });
+    const skipForks = resolveOrganizationSkipForks({
+      orgOverrides: organization.mirrorOverrides,
+      config,
+    });
+    const plan = partitionOrganizationRepositories(rows, { skipForks });
+
+    // Claim the organization before answering. The response already reports
+    // it as mirroring, and the background run starts a tick later, so a
+    // second request must not be able to slip through the gap. A claim that
+    // matches no row means another request took it in the meantime.
+    const claimed = await db
+      .update(organizations)
+      .set({ status: repoStatusEnum.parse("mirroring"), updatedAt: new Date() })
+      .where(
+        and(
+          eq(organizations.id, orgId),
+          eq(organizations.userId, userId),
+          notInArray(organizations.status, ["mirroring", "ignored"])
+        )
+      )
+      .returning({ id: organizations.id });
+
+    if (claimed.length === 0) {
+      return refusal(`Organization ${organization.name} is already being processed.`, 409);
+    }
+
+    // Fire the sync without blocking the response, the way mirror-org does.
+    setTimeout(() => {
+      runOrganizationSync({ config, userId, organization }).catch((error) => {
+        console.error(`Organization sync for ${organization.name} ended unexpectedly:`, error);
+      });
+    }, 0);
+
+    const responsePayload: SyncOrgResponse = {
+      success: true,
+      message: "Sync job started.",
+      organization: {
+        ...organization,
+        // The claim above already wrote this, so the card can show the
+        // spinner without waiting for the next fetch.
+        status: "mirroring",
+        membershipRole: organization.membershipRole as MembershipRole,
+        lastMirrored: organization.lastMirrored ?? undefined,
+        errorMessage: organization.errorMessage ?? undefined,
+        // The row keeps these null when they were never counted; the shared
+        // Organization type has them optional.
+        publicRepositoryCount: organization.publicRepositoryCount ?? undefined,
+        privateRepositoryCount: organization.privateRepositoryCount ?? undefined,
+        forkRepositoryCount: organization.forkRepositoryCount ?? undefined,
+      },
+      queued: {
+        mirror: plan.toMirror.length,
+        sync: plan.toSync.length,
+        skipped: plan.skipped.length,
+      },
+    };
+
+    return new Response(JSON.stringify(responsePayload), {
+      status: 200,
+      headers: { "Content-Type": "application/json" },
+    });
+  } catch (error) {
+    return createSecureErrorResponse(error, "organization sync", 500);
+  }
+};

src/types/sync.ts

@@ -1,4 +1,4 @@
-import type { Repository } from "@/lib/db/schema";
+import type { Organization, Repository } from "@/lib/db/schema";
 
 export interface SyncRepoRequest {
   userId: string;
@@ -12,6 +12,28 @@ export interface SyncRepoResponse {
   repositories: Repository[];
 }
 
+export interface SyncOrgRequest {
+  orgId: string;
+}
+
+/** What a manual organization sync put in the queue when it answered. */
+export interface SyncOrgQueuedCounts {
+  /** Repositories that were never mirrored and get an initial mirror. */
+  mirror: number;
+  /** Repositories already on the destination that get a refresh. */
+  sync: number;
+  /** Repositories the run leaves alone (in flight, ignored, being deleted). */
+  skipped: number;
+}
+
+export interface SyncOrgResponse {
+  success: boolean;
+  error?: string;
+  message?: string;
+  organization?: Organization;
+  queued?: SyncOrgQueuedCounts;
+}
+
 export interface ScheduleSyncRepoRequest {
   userId: string;
 }