fix: sweep every page of stale projects, fail the release job on failed publishes

.changeset/config.json

@@ -9,5 +9,5 @@
   "baseBranch": "main",
   "updateInternalDependencies": "patch",
   "ignore": ["create-db-worker", "claim-db-worker"],
-  "linked": [["create-db", "create-pg", "create-postgres"]]
+  "fixed": [["create-db", "create-pg", "create-postgres"]]
 }

.github/workflows/release.yml

@@ -72,23 +72,58 @@ jobs:
                   echo "Using CLAIM_DB_WORKER_URL=${CLAIM_DB_WORKER_URL}"
                   echo "Using POSTHOG_API_HOST=${POSTHOG_API_HOST}"
 
+                  # create-pg and create-postgres pin the create-db version they were
+                  # published with, so the three CLIs must move to a new version together
+                  versions=""
+                  for pkg in ${{ env.WORKSPACES }}; do
+                    versions="$versions $pkg@$(node -p "require('./${{ env.PACKAGES_DIR }}/$pkg/package.json').version")"
+                  done
+                  if [ "$(echo $versions | tr ' ' '\n' | sed 's/.*@//' | sort -u | wc -l)" -ne 1 ]; then
+                    echo "::error::The CLIs must share one version, got:$versions"
+                    exit 1
+                  fi
+
                   for pkg in ${{ env.WORKSPACES }}; do
-                    echo "Publishing $pkg to npm..."
                     cd "${{ env.PACKAGES_DIR }}/$pkg"
                     export POSTHOG_API_HOST="${POSTHOG_API_HOST}"
                     export POSTHOG_API_KEY="${POSTHOG_API_KEY}"
                     export CREATE_DB_WORKER_URL="${CREATE_DB_WORKER_URL}"
                     export CLAIM_DB_WORKER_URL="${CLAIM_DB_WORKER_URL}"
 
-                    # First try to publish
-                    if ! pnpm publish --access public --no-git-checks; then
-                      echo "Publish failed, trying to bump version and retry..."
-                      npm version patch --no-git-tag-version
-                      pnpm publish --access public --no-git-checks || echo "Publish failed again for $pkg"
+                    version=$(node -p "require('./package.json').version")
+
+                    # Merges that don't bump the version have nothing to publish
+                    if npm view "$pkg@$version" version >/dev/null 2>&1; then
+                      echo "$pkg@$version is already on npm, skipping."
+                    else
+                      echo "Publishing $pkg@$version to npm..."
+                      pnpm publish --access public --no-git-checks
                     fi
                     cd - >/dev/null
                   done
 
+            - name: ✅ Verify Published Versions
+              run: |
+                  for pkg in ${{ env.WORKSPACES }}; do
+                    version=$(node -p "require('./${{ env.PACKAGES_DIR }}/$pkg/package.json').version")
+
+                    # The registry can take a few seconds to serve a new version
+                    published=""
+                    for attempt in 1 2 3 4 5 6; do
+                      published=$(npm view "$pkg@$version" version 2>/dev/null || true)
+                      if [ "$published" = "$version" ]; then
+                        break
+                      fi
+                      sleep 10
+                    done
+
+                    if [ "$published" != "$version" ]; then
+                      echo "::error::$pkg@$version is not on npm"
+                      exit 1
+                    fi
+                    echo "$pkg@$version is on npm."
+                  done
+
             - name: 📝 Ensure Changeset Exists
               run: |
                   if [ -z "$(ls -A .changeset 2>/dev/null)" ]; then

apps/create-db-worker/src/delete-stale-workflow.ts

@@ -20,35 +20,63 @@ type ProjectsResponse = {
 	};
 };
 
+const PROJECTS_PAGE_SIZE = 1000;
+const STALE_AFTER_MS = 24 * 60 * 60 * 1000;
+
 export class DeleteStaleProjectsWorkflow extends WorkflowEntrypoint<Env, Params> {
 	async run(event: WorkflowEvent<Params>, step: WorkflowStep): Promise<void> {
-		const res = await step.do('fetch-projects', async () => {
-			const response = await fetch('https://api.prisma.io/v1/projects?limit=1000', {
-				method: 'GET',
-				headers: {
-					'Content-Type': 'application/json',
-					Authorization: `Bearer ${this.env.INTEGRATION_TOKEN}`,
-				},
+		const projects: Project[] = [];
+		const seenCursors = new Set<string>();
+		let cursor: string | null = null;
+		let page = 0;
+
+		do {
+			const pageCursor: string | null = cursor;
+			const res: string = await step.do(`fetch-projects-${page}`, async () => {
+				const url = new URL('https://api.prisma.io/v1/projects');
+				url.searchParams.set('limit', String(PROJECTS_PAGE_SIZE));
+				if (pageCursor) {
+					url.searchParams.set('cursor', pageCursor);
+				}
+
+				const response = await fetch(url, {
+					method: 'GET',
+					headers: {
+						'Content-Type': 'application/json',
+						Authorization: `Bearer ${this.env.INTEGRATION_TOKEN}`,
+					},
+				});
+
+				if (!response.ok) {
+					throw new Error(`Failed to fetch projects: ${response.statusText}`);
+				}
+
+				const data = await response.text();
+				return data;
 			});
 
-			if (!response.ok) {
-				throw new Error(`Failed to fetch projects: ${response.statusText}`);
-			}
+			const body: ProjectsResponse = JSON.parse(res);
+			projects.push(...body.data);
 
-			const data = await response.text();
-			return data;
-		});
+			const { hasMore, nextCursor } = body.pagination;
+			if (hasMore && (!nextCursor || seenCursors.has(nextCursor))) {
+				throw new Error(`Projects API reported more pages without a new cursor after page ${page}`);
+			}
+			cursor = hasMore ? nextCursor : null;
+			if (cursor) {
+				seenCursors.add(cursor);
+			}
+			page += 1;
+		} while (cursor);
 
-		const projects: ProjectsResponse = JSON.parse(res);
 		const now = Date.now();
-		const twentyFourHours = 24 * 60 * 60 * 1000;
 
-		const staleProjects = projects.data.filter((project) => {
+		const staleProjects = projects.filter((project) => {
 			const createdAt = new Date(project.createdAt).getTime();
-			return now - createdAt > twentyFourHours;
+			return now - createdAt > STALE_AFTER_MS;
 		});
 
-		console.log(`Total projects: ${projects.data.length}, Stale projects: ${staleProjects.length}`);
+		console.log(`Total projects: ${projects.length}, Stale projects: ${staleProjects.length}`);
 
 		for (const project of staleProjects) {
 			await step.do(`delete-project-${project.id}`, async () => {

apps/create-db-worker/test/delete-stale-workflow.test.ts

@@ -0,0 +1,114 @@
+import { describe, it, expect, vi, beforeEach } from 'vitest';
+import { DeleteStaleProjectsWorkflow } from '../src/delete-stale-workflow';
+
+const HOUR_MS = 60 * 60 * 1000;
+
+function hoursAgo(hours: number) {
+	return new Date(Date.now() - hours * HOUR_MS).toISOString();
+}
+
+function page(data: { id: string; createdAt: string }[], nextCursor: string | null) {
+	return new Response(
+		JSON.stringify({
+			data: data.map((project) => ({ ...project, name: project.id })),
+			pagination: { hasMore: nextCursor !== null, nextCursor },
+		}),
+	);
+}
+
+async function runWorkflow() {
+	const stepNames: string[] = [];
+	const step = {
+		do: async (name: string, callback: () => Promise<unknown>) => {
+			stepNames.push(name);
+			return callback();
+		},
+	};
+	// workerd only constructs a WorkflowEntrypoint itself, so run the method on a stand-in `this`
+	await DeleteStaleProjectsWorkflow.prototype.run.call(
+		{ env: { INTEGRATION_TOKEN: 'test-token' } } as any,
+		{} as any,
+		step as any,
+	);
+	return stepNames;
+}
+
+describe('DeleteStaleProjectsWorkflow', () => {
+	let fetchMock: ReturnType<typeof vi.fn>;
+
+	function requestsTo(method: string) {
+		return fetchMock.mock.calls.filter(([, init]) => init?.method === method).map(([url]) => new URL(url));
+	}
+
+	beforeEach(() => {
+		fetchMock = vi.fn();
+		vi.spyOn(globalThis, 'fetch').mockImplementation(fetchMock);
+	});
+
+	it('walks every page and deletes stale projects from all of them', async () => {
+		fetchMock
+			.mockResolvedValueOnce(
+				page(
+					[
+						{ id: 'stale-1', createdAt: hoursAgo(30) },
+						{ id: 'fresh-1', createdAt: hoursAgo(2) },
+					],
+					'fresh-1',
+				),
+			)
+			.mockResolvedValueOnce(page([{ id: 'stale-2', createdAt: hoursAgo(25) }], null))
+			.mockImplementation(async () => new Response(null, { status: 200 }));
+
+		const stepNames = await runWorkflow();
+
+		const lists = requestsTo('GET');
+		expect(lists).toHaveLength(2);
+		expect(lists[0].searchParams.get('cursor')).toBeNull();
+		expect(lists[1].searchParams.get('cursor')).toBe('fresh-1');
+
+		expect(requestsTo('DELETE').map((url) => url.pathname)).toEqual([
+			'/v1/projects/stale-1',
+			'/v1/projects/stale-2',
+		]);
+
+		expect(new Set(stepNames).size).toBe(stepNames.length);
+	});
+
+	it('stops after one page when there are no more projects', async () => {
+		fetchMock.mockResolvedValueOnce(page([{ id: 'fresh-1', createdAt: hoursAgo(1) }], null));
+
+		await runWorkflow();
+
+		expect(requestsTo('GET')).toHaveLength(1);
+		expect(requestsTo('DELETE')).toHaveLength(0);
+	});
+
+	it('fails when the API reports more pages without advancing the cursor', async () => {
+		fetchMock
+			.mockResolvedValueOnce(page([{ id: 'stale-1', createdAt: hoursAgo(30) }], 'stale-1'))
+			.mockResolvedValueOnce(page([{ id: 'stale-1', createdAt: hoursAgo(30) }], 'stale-1'));
+
+		await expect(runWorkflow()).rejects.toThrow('without a new cursor');
+		expect(requestsTo('DELETE')).toHaveLength(0);
+	});
+
+	it('fails when the API reports more pages with no cursor', async () => {
+		fetchMock.mockResolvedValueOnce(
+			new Response(
+				JSON.stringify({
+					data: [{ id: 'stale-1', name: 'stale-1', createdAt: hoursAgo(30) }],
+					pagination: { hasMore: true, nextCursor: null },
+				}),
+			),
+		);
+
+		await expect(runWorkflow()).rejects.toThrow('without a new cursor');
+		expect(requestsTo('DELETE')).toHaveLength(0);
+	});
+
+	it('fails when a page cannot be fetched', async () => {
+		fetchMock.mockResolvedValueOnce(new Response(null, { status: 500, statusText: 'Server Error' }));
+
+		await expect(runWorkflow()).rejects.toThrow('Failed to fetch projects');
+	});
+});