diff --git a/src/components/organizations/Organization.tsx b/src/components/organizations/Organization.tsx index 7f943c7..705be34 100644 --- a/src/components/organizations/Organization.tsx +++ b/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("/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} diff --git a/src/components/organizations/OrganizationsList.tsx b/src/components/organizations/OrganizationsList.tsx index 7a81bf2..83434ae 100644 --- a/src/components/organizations/OrganizationsList.tsx +++ b/src/components/organizations/OrganizationsList.tsx @@ -48,6 +48,12 @@ interface OrganizationListProps { filter: FilterParams; setFilter: (filter: FilterParams) => void; onMirror: ({ orgId }: { orgId: string }) => Promise; + /** + * 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; onIgnore?: ({ orgId, ignore }: { orgId: string; ignore: boolean }) => Promise; loadingOrgIds: Set; onAddOrganization?: () => void; @@ -145,6 +151,7 @@ export function OrganizationList({ filter, setFilter, onMirror, + onSync, onIgnore, loadingOrgIds, onAddOrganization, @@ -726,6 +733,15 @@ export function OrganizationList({ + {onSync && (org.status === "mirrored" || org.status === "failed") && ( + <> + org.id && onSync({ orgId: org.id })}> + + Sync Organization + + + + )} setOverridesTarget(org)}> Mirror Options @@ -877,6 +893,15 @@ export function OrganizationList({ + {onSync && (org.status === "mirrored" || org.status === "failed") && ( + <> + org.id && onSync({ orgId: org.id })}> + + Sync Organization + + + + )} setOverridesTarget(org)}> Mirror Options diff --git a/src/lib/organization-discovery.ts b/src/lib/organization-discovery.ts new file mode 100644 index 0000000..034e4d7 --- /dev/null +++ b/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 & { 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; + /** Named in the log lines, e.g. "scheduled sync". */ + phase: string; + logPrefix?: string; +}): Promise { + // 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, + }; +} diff --git a/src/lib/organization-sync.ts b/src/lib/organization-sync.ts new file mode 100644 index 0000000..a2fe3c1 --- /dev/null +++ b/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 { + 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; + repository: Repository; + octokit: Octokit | null; + sourceUsername?: string; +}): Promise { + 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 & { id: string }; + userId: string; + organization: SyncableOrganization; +}): Promise { + 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; + } +} diff --git a/src/lib/scheduler-service.ts b/src/lib/scheduler-service.ts index 1571cf6..e9fcd9f 100644 --- a/src/lib/scheduler-service.ts +++ b/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 - * 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 + * Re-discover organization repositories: 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 - * idempotent). Unpinned organizations and pins whose source is gone keep - * the legacy DB-only behavior — no behavior change for existing users. - * - * Repositories missing from the listing are deliberately NOT removed: - * a tokenless listing only shows public repositories, so cleanup stays - * membership-based (repository-cleanup-service). + * the repositories it does not have yet. The per-organization work lives in + * rediscoverRepositoriesForOrganization, which the manual organization sync + * route shares. */ async function rediscoverOrganizationRepositories( config: any, userId: string, phase: 'scheduled sync' | 'auto-start' ): Promise { - const { listSources, findSourceForOrganization } = await import('@/lib/sources'); - const { createSourceProviderFromSource } = await import('@/lib/source-providers'); + const { listSources } = await import('@/lib/sources'); 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); - if (!source) continue; - - const sourceProvider = createSourceProviderFromSource(source, { userId }); - const orgRepos = await sourceProvider.listOrganizationRepositories(org.name); - - // The organization's fork policy (override -> global skipForks), the - // 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)); + await rediscoverRepositoriesForOrganization({ + config, + userId, + organization: org, + sources, + existingRepoKeys, + phase, + }); } catch (orgError) { console.error(`[Scheduler] Failed to re-discover repositories for organization ${org.name} for user ${userId} during ${phase}:`, orgError); } diff --git a/src/pages/api/job/sync-org.test.ts b/src/pages/api/job/sync-org.test.ts new file mode 100644 index 0000000..4135e43 --- /dev/null +++ b/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(); +const syncSkips = new Set(); +/** + * 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) => { + 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) { + 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 { + 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); + }); +}); diff --git a/src/pages/api/job/sync-org.ts b/src/pages/api/job/sync-org.ts new file mode 100644 index 0000000..fb1a025 --- /dev/null +++ b/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 : ", 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); + } +}; diff --git a/src/types/sync.ts b/src/types/sync.ts index 24c14b2..18c7113 100644 --- a/src/types/sync.ts +++ b/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; }