feat(triggers): adopt connector-owned sources

apps/server/migrations/0054_source_references.sql

@@ -0,0 +1,12 @@
+-- Existing Source snapshots need an explicit deployment decision before this upgrade.
+CREATE TEMP TABLE source_reference_preflight (empty INTEGER CONSTRAINT legacy_sources_require_explicit_migration CHECK (empty = 1));
+INSERT INTO source_reference_preflight SELECT NOT EXISTS (SELECT 1 FROM event_sources);
+DROP TABLE source_reference_preflight;
+
+ALTER TABLE event_sources ADD COLUMN connector_json TEXT CHECK (connector_json IS NULL OR json_valid(connector_json));
+ALTER TABLE event_sources ADD COLUMN pending_json TEXT CHECK (pending_json IS NULL OR json_valid(pending_json));
+CREATE TABLE event_source_namespace (
+  id INTEGER PRIMARY KEY CHECK (id = 1),
+  value TEXT NOT NULL
+) STRICT;
+INSERT INTO event_source_namespace VALUES (1, lower(hex(randomblob(16))));

apps/server/node/application/event-source-control.ts

@@ -50,10 +50,10 @@ export class EventSourceControl {
     return this.eventSources.update(sourceId, input, signal)
   }
 
-  deleteEventSource(actorId: string, sourceId: string, revision: number) {
+  deleteEventSource(actorId: string, sourceId: string, revision: number, signal = AbortSignal.timeout(30_000)) {
     this.access.requireAdmin(actorId)
 
     if (this.eventSources == null) throw new ControlError(controlErrorCode.eventSourceInvalid, 'Event sources are unavailable.')
-    this.eventSources.delete(sourceId, revision)
+    return this.eventSources.delete(sourceId, revision, signal)
   }
 }

apps/server/node/deployment/connector-contract.ts

@@ -78,10 +78,12 @@ export type ConnectorErrorCode =
 
 export class ConnectorTaskError extends Error {
   readonly code: ConnectorErrorCode
+  readonly status?: number
 
-  constructor(code: ConnectorErrorCode, message: string) {
+  constructor(code: ConnectorErrorCode, message: string, status?: number) {
     super(message)
     this.code = code
+    this.status = status
     this.name = 'ConnectorTaskError'
   }
 

apps/server/node/deployment/connector-transport.ts

@@ -18,13 +18,11 @@ interface GetOptions {
 
 export class RequestError extends ConnectorTaskError {
   readonly reason: string
-  readonly status: number | undefined
 
   constructor(cause: ConnectorTaskError, reason: string, status: number | undefined) {
-    super(cause.code, cause.message)
+    super(cause.code, cause.message, status)
     this.cause = cause
     this.reason = reason
-    this.status = status
   }
 }
 

apps/server/node/deployment/connector.ts

@@ -454,18 +454,20 @@ export class ConnectorClient implements ConnectorHost {
       (access.scope != 'trigger' || access.triggerId != triggerId || access.providerId != provider || access.connectionId != connectionId)
     )
       throw accessInvalid()
-    const bindings = await this.#providerAccessBindings(access, provider, signal, true)
+    const sourceIngress = request.operation === 'sourceReference' && (request.request.action === 'delete' || request.request.action === 'route')
+    if (sourceIngress && (typeof access !== 'object' || access.scope !== 'catalog' || access.source !== 'operator')) throw accessInvalid()
+    const bindings = sourceIngress ? null : await this.#providerAccessBindings(access, provider, signal, true)
     if (bindings != null && !bindings.some((candidate) => candidate.appId == connectionId && providerAccessAllowsTrigger(candidate, triggerId)))
       throw accessInvalid()
     const teamId = connectorTeamId(access)
-    await this.#assertConnection(connectionId, provider, signal, teamId)
+    if (!sourceIngress) await this.#assertConnection(connectionId, provider, signal, teamId)
     const response = await this.#transport.request(
       'trigger.execute',
-      `v1/providers/${encodeURIComponent(provider)}/triggers/${encodeURIComponent(triggerId)}/execute`,
+      `v1/providers/${encodeURIComponent(provider)}/triggers/${encodeURIComponent(triggerId)}/execute?appId=${encodeURIComponent(connectionId)}`,
       {
         method: 'POST',
         body: JSON.stringify(request),
-        headers: { 'content-type': 'application/json', 'x-oo-connector-app-id': connectionId },
+        headers: { 'content-type': 'application/json' },
       },
       signal,
       { fields: { connectionId, provider, triggerId }, teamId },
@@ -485,6 +487,8 @@ export class ConnectorClient implements ConnectorHost {
         'connector.input-invalid',
         'Webhook creation has an uncertain outcome. Check the existing remote webhook before retrying; do not create another event source.',
       )
+    if (request.operation === 'sourceReference' && response.status === 409 && record(response.value) && response.value.errorCode === 'invalid_input')
+      throw new ConnectorTaskError('connector.input-invalid', 'The Source changed. Refresh its configuration before retrying.', 409)
     if (record(response.value) && ['invalid_input', 'trigger_not_found'].includes(String(response.value.errorCode)) && response.status != 409)
       throw new ConnectorTaskError('connector.input-invalid', 'The Trigger operation configuration is invalid.')
     if (