feat(web): add POST /api/connections/{id}/sync to trigger a connection sync over the public API

CHANGELOG.md

@@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+- `POST /api/connections/{id}/sync` enqueues a new `ConnectionSyncJob` for a connection in the caller's org; the caller must be an OWNER, cross-org access returns 404, and the response is `{ jobId }` so on-call scripts can force a re-sync without going through the web UI. [#1520](https://github.com/sourcebot-dev/sourcebot/pull/1520)
+
 ### Fixed
 - Upgraded `seroval` to `^1.5.6`. [#1508](https://github.com/sourcebot-dev/sourcebot/pull/1508)
 

docs/api-reference/sourcebot-public.openapi.json

@@ -537,6 +537,18 @@
           "status"
         ]
       },
+      "PublicSyncConnectionResponse": {
+        "type": "object",
+        "properties": {
+          "jobId": {
+            "type": "string",
+            "description": "The id of the new ConnectionSyncJob row. Poll the connection detail endpoint to see when the job transitions from PENDING/IN_PROGRESS to COMPLETED/FAILED."
+          }
+        },
+        "required": [
+          "jobId"
+        ]
+      },
       "PublicFileSourceResponse": {
         "type": "object",
         "properties": {
@@ -1479,6 +1491,90 @@
         }
       }
     },
+    "/api/connections/{id}/sync": {
+      "post": {
+        "operationId": "triggerConnectionSync",
+        "tags": [
+          "System"
+        ],
+        "summary": "Trigger a connection sync",
+        "description": "Enqueues a new ConnectionSyncJob row for the connection in the caller's org. The job runs asynchronously; the response returns the new job id. The caller must be an OWNER in the org. Cross-org access returns 404 (not 403) to avoid leaking the existence of connections in other orgs.",
+        "parameters": [
+          {
+            "schema": {
+              "type": "integer",
+              "minimum": 0,
+              "exclusiveMinimum": true
+            },
+            "required": true,
+            "name": "id",
+            "in": "path"
+          }
+        ],
+        "responses": {
+          "202": {
+            "description": "Sync job enqueued. Poll `GET /api/connections/{id}` to see the job progress.",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/PublicSyncConnectionResponse"
+                }
+              }
+            }
+          },
+          "400": {
+            "description": "Invalid `id` path parameter.",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/PublicApiServiceError"
+                }
+              }
+            }
+          },
+          "401": {
+            "description": "Not authenticated.",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/PublicApiServiceError"
+                }
+              }
+            }
+          },
+          "403": {
+            "description": "Caller is not an OWNER in the org.",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/PublicApiServiceError"
+                }
+              }
+            }
+          },
+          "404": {
+            "description": "Connection not found in the org.",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/PublicApiServiceError"
+                }
+              }
+            }
+          },
+          "500": {
+            "description": "Worker is unreachable or returned an unexpected response.",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/PublicApiServiceError"
+                }
+              }
+            }
+          }
+        }
+      }
+    },
     "/api/source": {
       "get": {
         "operationId": "getFileSource",

docs/docs.json

@@ -217,7 +217,8 @@
                         "icon": "server",
                         "pages": [
                             "GET /api/version",
-                            "GET /api/health"
+                            "GET /api/health",
+                            "POST /api/connections/{id}/sync"
                         ]
                     }
                 ]

packages/web/src/app/api/(server)/connections/[id]/sync/route.test.ts

@@ -0,0 +1,193 @@
+import { beforeEach, describe, expect, test, vi } from 'vitest';
+import { NextRequest } from 'next/server';
+import { OrgRole } from '@sourcebot/db';
+
+type AuthContext = {
+    user: { id: string };
+    org: { id: number };
+    role: OrgRole;
+    prisma: unknown;
+};
+
+const mocks = vi.hoisted(() => ({
+    authContext: undefined as AuthContext | undefined,
+    findFirst: vi.fn(),
+    workerResponse: undefined as { ok: boolean; status: number; json: () => Promise<unknown> } | undefined,
+}));
+
+vi.mock('server-only', () => ({}));
+
+vi.mock('@sourcebot/db', () => ({
+    OrgRole: { MEMBER: 'MEMBER', OWNER: 'OWNER' },
+}));
+
+vi.mock('@sourcebot/shared', () => ({
+    env: { WORKER_API_URL: 'http://worker.example.com' },
+    createLogger: () => ({
+        debug: vi.fn(),
+        info: vi.fn(),
+        warn: vi.fn(),
+        error: vi.fn(),
+    }),
+}));
+
+vi.mock('@/lib/posthog', () => ({
+    captureEvent: vi.fn(),
+}));
+
+vi.mock('@/middleware/withAuth', () => ({
+    withAuth: vi.fn(async (callback: (ctx: AuthContext) => unknown) => {
+        if (!mocks.authContext) {
+            return { statusCode: 401, errorCode: 'NOT_AUTHENTICATED', message: 'Not authenticated' };
+        }
+        return callback(mocks.authContext);
+    }),
+}));
+
+vi.mock('@sentry/nextjs', () => ({
+    captureException: vi.fn(),
+    captureRequestError: vi.fn(),
+}));
+
+vi.mock('@opentelemetry/sdk-trace-base', () => ({
+    getEnv: () => ({}),
+    TraceIdRatioBasedSampler: vi.fn(),
+    ParentBasedSampler: vi.fn(),
+    AlwaysOnSampler: vi.fn(),
+    AlwaysOffSampler: vi.fn(),
+}));
+
+const setAuth = (opts: {
+    role?: OrgRole;
+    connectionExists?: boolean;
+}) => {
+    mocks.findFirst.mockImplementation(async () => opts.connectionExists ? { id: 1 } : null);
+    mocks.authContext = {
+        user: { id: 'user_1' },
+        org: { id: 1 },
+        role: opts.role ?? OrgRole.OWNER,
+        prisma: { connection: { findFirst: mocks.findFirst } },
+    };
+};
+
+const makeRequest = (id: string): NextRequest =>
+    new NextRequest(`http://localhost/api/connections/${id}/sync`, { method: 'POST' });
+
+const { POST } = await import('./route');
+
+// Mock the global fetch that the action uses to call the worker.
+const originalFetch = globalThis.fetch;
+beforeEach(() => {
+    vi.restoreAllMocks();
+    globalThis.fetch = vi.fn(async () => {
+        if (!mocks.workerResponse) {
+            throw new Error('workerResponse not set');
+        }
+        return new Response(JSON.stringify(await mocks.workerResponse.json()), {
+            status: mocks.workerResponse.status,
+            headers: { 'Content-Type': 'application/json' },
+        });
+    }) as typeof fetch;
+    mocks.authContext = undefined;
+    mocks.workerResponse = undefined;
+});
+
+// Restore the original fetch when the suite finishes so other test
+// files are not affected.
+import { afterAll } from 'vitest';
+afterAll(() => {
+    globalThis.fetch = originalFetch;
+});
+
+describe('POST /api/connections/:id/sync', () => {
+    test('returns 401 when no authenticated user', async () => {
+        mocks.authContext = undefined;
+        const response = await POST(makeRequest('1'), { params: Promise.resolve({ id: '1' }) });
+        expect(response.status).toBe(401);
+    });
+
+    test('returns 403 when the caller is a MEMBER (not OWNER)', async () => {
+        setAuth({ role: OrgRole.MEMBER, connectionExists: true });
+        const response = await POST(makeRequest('1'), { params: Promise.resolve({ id: '1' }) });
+        expect(response.status).toBe(403);
+        // The worker must not be called when the caller is a MEMBER.
+        expect(globalThis.fetch).not.toHaveBeenCalled();
+    });
+
+    test('returns 400 when id is not a positive integer', async () => {
+        setAuth({ role: OrgRole.OWNER, connectionExists: false });
+        const response = await POST(makeRequest('abc'), { params: Promise.resolve({ id: 'abc' }) });
+        expect(response.status).toBe(400);
+        expect(globalThis.fetch).not.toHaveBeenCalled();
+    });
+
+    test('returns 400 when id is zero or negative', async () => {
+        setAuth({ role: OrgRole.OWNER, connectionExists: false });
+        const response = await POST(makeRequest('0'), { params: Promise.resolve({ id: '0' }) });
+        expect(response.status).toBe(400);
+        expect(globalThis.fetch).not.toHaveBeenCalled();
+    });
+
+    test('returns 404 when the connection does not exist in the org', async () => {
+        setAuth({ role: OrgRole.OWNER, connectionExists: false });
+        const response = await POST(makeRequest('999'), { params: Promise.resolve({ id: '999' }) });
+        expect(response.status).toBe(404);
+        expect(globalThis.fetch).not.toHaveBeenCalled();
+    });
+
+    test('returns 404 (not 403) for cross-org access to avoid leaking existence', async () => {
+        // The mock returns null for `connectionExists: false` regardless
+        // of which id is requested, simulating the case where the
+        // connection lives in a different org and is therefore invisible
+        // to the current org. The action scopes by `where: { id, orgId }`
+        // so cross-org ids collapse to "not found", same as the detail
+        // endpoint.
+        setAuth({ role: OrgRole.OWNER, connectionExists: false });
+        const response = await POST(makeRequest('42'), { params: Promise.resolve({ id: '42' }) });
+        expect(response.status).toBe(404);
+        // Assert that the action scoped the query by orgId. A future
+        // change that drops the orgId filter would still pass the 404
+        // assertion (mock returns null regardless of args) but would
+        // expose cross-org data in production.
+        expect(mocks.findFirst).toHaveBeenCalledWith(
+            expect.objectContaining({ where: { id: 42, orgId: 1 } }),
+        );
+        expect(globalThis.fetch).not.toHaveBeenCalled();
+    });
+
+    test('returns 202 with the new jobId on the happy path', async () => {
+        setAuth({ role: OrgRole.OWNER, connectionExists: true });
+        mocks.workerResponse = {
+            ok: true,
+            status: 200,
+            json: async () => ({ jobId: 'job_abc' }),
+        };
+
+        const response = await POST(makeRequest('1'), { params: Promise.resolve({ id: '1' }) });
+        const body = await response.json();
+
+        expect(response.status).toBe(202);
+        expect(body).toEqual({ jobId: 'job_abc' });
+
+        // The action called the worker with the connectionId in the body.
+        expect(globalThis.fetch).toHaveBeenCalledWith(
+            'http://worker.example.com/api/sync-connection',
+            expect.objectContaining({
+                method: 'POST',
+                body: JSON.stringify({ connectionId: 1 }),
+            }),
+        );
+    });
+
+    test('returns 500 when the worker returns a non-ok response', async () => {
+        setAuth({ role: OrgRole.OWNER, connectionExists: true });
+        mocks.workerResponse = {
+            ok: false,
+            status: 500,
+            json: async () => ({ error: 'worker exploded' }),
+        } as unknown as { ok: boolean; status: number; json: () => Promise<unknown> };
+
+        const response = await POST(makeRequest('1'), { params: Promise.resolve({ id: '1' }) });
+        expect(response.status).toBe(500);
+    });
+});

packages/web/src/app/api/(server)/connections/[id]/sync/route.ts

@@ -0,0 +1,32 @@
+'use server';
+
+import { NextRequest } from 'next/server';
+import { z } from 'zod';
+
+import { triggerConnectionSync } from '@/features/workerApi/actions';
+import { apiHandler } from '@/lib/apiHandler';
+import { queryParamsSchemaValidationError, serviceErrorResponse } from '@/lib/serviceError';
+import { isServiceError } from '@/lib/utils';
+
+const connectionIdParamSchema = z.coerce.number().int().positive();
+
+// eslint-disable-next-line authz/require-auth-wrapper -- delegates to triggerConnectionSync() which calls withAuth + withMinimumOrgRole(OWNER)
+export const POST = apiHandler(async (
+    _request: NextRequest,
+    { params }: { params: Promise<{ id: string }> },
+) => {
+    const { id: rawId } = await params;
+    const idParsed = connectionIdParamSchema.safeParse(rawId);
+    if (!idParsed.success) {
+        return serviceErrorResponse(queryParamsSchemaValidationError(idParsed.error));
+    }
+    const id = idParsed.data;
+
+    const result = await triggerConnectionSync(id);
+
+    if (isServiceError(result)) {
+        return serviceErrorResponse(result);
+    }
+
+    return Response.json({ jobId: result.jobId }, { status: 202 });
+});