feat: export Catalog Studio assets and SQL to catalog repository

src/lib/api-handler.ts

@@ -1,9 +1,15 @@
-import { type NextRequest, NextResponse } from "next/server";
+import { after, type NextRequest, NextResponse } from "next/server";
 import { logAuthorizationEvent } from "@/lib/admin/authorization-events";
 import { validateCsrfToken } from "@/lib/foundation/security";
 import { canAccess, getApiAdminContext } from "@/lib/permissions";
 import { logServerError } from "@/lib/server-log";
 
+import {
+	catalogExportEnabled,
+	catalogExportQueue,
+	isCatalogMutation,
+} from "@/lib/services/catalog-git-queue";
+
 const MUTATING_METHODS = new Set(["POST", "PUT", "PATCH", "DELETE"]);
 const MAX_BODY_BYTES = 10 * 1024 * 1024; // 10 MB
 
@@ -80,9 +86,42 @@ export function withAdmin(
 				{ status: 403 },
 			);
 		}
+		let finishExport: (() => Promise<void>) | undefined;
 		try {
-			return await handler(request, context, routeContext);
+			if (
+				catalogExportEnabled() &&
+				isCatalogMutation(request.method, request.nextUrl.pathname)
+			) {
+				finishExport = await catalogExportQueue().begin();
+			}
+			const response = await handler(request, context, routeContext);
+			if (
+				finishExport &&
+				response.body &&
+				response.headers.get("content-type")?.includes("text/event-stream")
+			) {
+				const [client, completion] = response.body.tee();
+				const finish = finishExport;
+				after(async () => {
+					const reader = completion.getReader();
+					try {
+						while (!(await reader.read()).done) {
+							/* Wait for all import results. */
+						}
+					} finally {
+						reader.releaseLock();
+						await finish();
+					}
+				});
+				return new Response(client, {
+					status: response.status,
+					headers: response.headers,
+				});
+			}
+			await finishExport?.();
+			return response;
 		} catch (error) {
+			await finishExport?.().catch(() => undefined);
 			logServerError("admin.api_failed", error, {
 				path: request.nextUrl.pathname,
 				userId: context.session.user.id,