feat: Webhooks

package.json

@@ -20,6 +20,7 @@
     "tool:restore": "tsx tools/restore.ts",
     "tool:pruneBackups": "tsx tools/pruneBackups.ts",
     "tool:cleanupSessions": "tsx tools/cleanupSessions.ts",
+    "tool:pruneWebhookLogs": "tsx tools/pruneWebhookLogs.ts",
     "tool:seed-mirror": "tsx tools/seedMirror.ts",
     "cli": "tsx src/cli/cli.ts",
     "test": "vitest run",

server.ts

@@ -30,9 +30,11 @@ import { rateLimitMiddleware } from '~/api/rate-limit.server'
 import _records from '~/api/records'
 import _schemas from '~/api/schemas'
 import _versions from '~/api/versions'
+import _webhooks from '~/api/webhooks'
 import { auth } from '~/lib/auth'
 import { getSessionUser } from '~/lib/auth.server'
 import { getMirrorConfig } from '~/lib/mirror-config'
+import { startWebhookBackgroundJobs } from '~/lib/webhooks.server'
 
 const isProd = process.env.NODE_ENV === 'production'
 let vite: ViteDevServer | undefined
@@ -229,11 +231,15 @@ const routes = app
   .route(...api('/api', './src/api/collections.ts', _collections))
   .route(...api('/api/collections', './src/api/versions.ts', _versions))
   .route(...api('/api/collections', './src/api/negotiate.ts', _negotiate))
+  .route(...api('/api/collections', './src/api/webhooks.ts', _webhooks))
   .route(...api('/api', './src/api/discussion.ts', _discussion))
   .route(...api('/api/organizations', './src/api/organizations.ts', _organizations))
 
 export type AppType = typeof routes
 
+// Retry sweep + delivery-log purge for webhooks (in-process, unref'd intervals)
+startWebhookBackgroundJobs()
+
 // --- Legacy API routes (not yet converted to subapp pattern) ---
 app.get('/api/health', health.check)
 

src/api/negotiate.ts

@@ -25,6 +25,11 @@ import {
   type SchemaEntry,
   stripToSchema,
 } from '../lib/version-helpers.server.js'
+import {
+  bumpTypeFromChanges,
+  dispatchDeliveries,
+  enqueueWebhookDeliveries,
+} from '../lib/webhooks.server.js'
 import { requireAuth, type AuthEnv } from './auth.server.js'
 
 const SESSION_TTL_MS = 10 * 60 * 1000
@@ -975,6 +980,27 @@ app.post(
       .set({ status: 'committed' })
       .where(eq(schema.negotiateSessions.id, sessionId))
 
+    // Fire webhooks for the new version — best-effort, never blocks/denies the 201.
+    try {
+      const deliveryIds = await enqueueWebhookDeliveries(
+        {
+          id: versionId!,
+          semver: sv.semver,
+          hash: versionHash,
+          major: sv.major,
+          minor: sv.minor,
+          patch: sv.patch,
+          recordCount: finalRecordHashes.length,
+          fileCount: session.fileHashes.length,
+        },
+        bumpTypeFromChanges(schemaChanged, recordsChanged),
+        session.collectionId,
+      )
+      dispatchDeliveries(deliveryIds)
+    } catch (err) {
+      console.error(`[webhooks] failed to enqueue for ${sv.semver}:`, err)
+    }
+
     return c.json(
       {
         semver: sv.semver,

src/api/versions.ts

@@ -23,6 +23,7 @@ import {
   resolveCollection,
   type SchemaEntry,
 } from '../lib/version-helpers.server.js'
+import { dispatchDeliveries, enqueueWebhookDeliveries } from '../lib/webhooks.server.js'
 import { type AuthEnv, requireAuth } from './auth.server.js'
 
 const MAX_METADATA_BYTES = 64 * 1024
@@ -953,6 +954,7 @@ const app = new Hono<AuthEnv>()
 
       const sv = deriveSemver(latest.semver, false, false, true)
 
+      let newVersionId: number | undefined
       await db.transaction(async (tx) => {
         const [version] = await tx
           .insert(schema.versions)
@@ -974,6 +976,8 @@ const app = new Hono<AuthEnv>()
           })
           .returning({ id: schema.versions.id })
 
+        newVersionId = version!.id
+
         if (schemaEntries.length > 0) {
           await tx.insert(schema.versionSchemas).values(
             schemaEntries.map((e) => ({
@@ -1007,6 +1011,27 @@ const app = new Hono<AuthEnv>()
           .where(eq(schema.collections.id, collection.id))
       })
 
+      // Fire webhooks for the metadata (patch) version — best-effort.
+      try {
+        const deliveryIds = await enqueueWebhookDeliveries(
+          {
+            id: newVersionId!,
+            semver: sv.semver,
+            hash: versionHash,
+            major: sv.major,
+            minor: sv.minor,
+            patch: sv.patch,
+            recordCount: latest.recordCount,
+            fileCount: latest.fileCount,
+          },
+          'patch',
+          collection.id,
+        )
+        dispatchDeliveries(deliveryIds)
+      } catch (err) {
+        console.error(`[webhooks] failed to enqueue for ${sv.semver}:`, err)
+      }
+
       return c.json({ semver: sv.semver, hash: versionHash, metadata: newMetadata }, 201)
     },
   )