feat(enterprise): add data drains for continuous export to S3 / webhook (#4440)

* feat(enterprise): add data drains for continuous export to S3 / webhook

* chore(data-drains): regenerate migration on top of staging + bump route baseline

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* docs(data-drains): clarify retention pairing is user-coupled, not enforced

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* fix(data-drains): preserve explicit forcePathStyle=false + reserve x-sim-signature

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* test(data-drains): drift guard ensures every webhook header is reserved

Asserts that any header buildHeaders writes is rejected when reused as a
custom signatureHeader. Adding a new metadata header without mirroring it
into RESERVED_SIGNATURE_HEADER_NAMES now fails CI.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>

packages/audit/src/types.ts

@@ -24,6 +24,13 @@ export const AuditAction = {
   CUSTOM_TOOL_UPDATED: 'custom_tool.updated',
   CUSTOM_TOOL_DELETED: 'custom_tool.deleted',
 
+  // Data Drains
+  DATA_DRAIN_CREATED: 'data_drain.created',
+  DATA_DRAIN_UPDATED: 'data_drain.updated',
+  DATA_DRAIN_DELETED: 'data_drain.deleted',
+  DATA_DRAIN_RAN: 'data_drain.ran',
+  DATA_DRAIN_TESTED: 'data_drain.tested',
+
   // Billing
   CREDIT_PURCHASED: 'credit.purchased',
 
@@ -194,6 +201,7 @@ export const AuditResourceType = {
   CREDENTIAL: 'credential',
   CREDENTIAL_SET: 'credential_set',
   CUSTOM_TOOL: 'custom_tool',
+  DATA_DRAIN: 'data_drain',
   DOCUMENT: 'document',
   ENVIRONMENT: 'environment',
   FILE: 'file',

packages/db/migrations/0204_powerful_medusa.sql

@@ -0,0 +1,50 @@
+CREATE TYPE "public"."data_drain_cadence" AS ENUM('hourly', 'daily');--> statement-breakpoint
+CREATE TYPE "public"."data_drain_destination" AS ENUM('s3', 'webhook');--> statement-breakpoint
+CREATE TYPE "public"."data_drain_run_status" AS ENUM('running', 'success', 'failed');--> statement-breakpoint
+CREATE TYPE "public"."data_drain_run_trigger" AS ENUM('cron', 'manual');--> statement-breakpoint
+CREATE TYPE "public"."data_drain_source" AS ENUM('workflow_logs', 'job_logs', 'audit_logs', 'copilot_chats', 'copilot_runs');--> statement-breakpoint
+CREATE TABLE "data_drain_runs" (
+	"id" text PRIMARY KEY NOT NULL,
+	"drain_id" text NOT NULL,
+	"status" "data_drain_run_status" NOT NULL,
+	"trigger" "data_drain_run_trigger" NOT NULL,
+	"started_at" timestamp DEFAULT now() NOT NULL,
+	"finished_at" timestamp,
+	"rows_exported" integer DEFAULT 0 NOT NULL,
+	"bytes_written" bigint DEFAULT 0 NOT NULL,
+	"cursor_before" text,
+	"cursor_after" text,
+	"error" text,
+	"locators" jsonb DEFAULT '[]'::jsonb NOT NULL
+);
+--> statement-breakpoint
+CREATE TABLE "data_drains" (
+	"id" text PRIMARY KEY NOT NULL,
+	"organization_id" text NOT NULL,
+	"name" text NOT NULL,
+	"source" "data_drain_source" NOT NULL,
+	"destination_type" "data_drain_destination" NOT NULL,
+	"destination_config" jsonb NOT NULL,
+	"destination_credentials" text NOT NULL,
+	"schedule_cadence" "data_drain_cadence" NOT NULL,
+	"enabled" boolean DEFAULT true NOT NULL,
+	"cursor" text,
+	"last_run_at" timestamp,
+	"last_success_at" timestamp,
+	"created_by" text NOT NULL,
+	"created_at" timestamp DEFAULT now() NOT NULL,
+	"updated_at" timestamp DEFAULT now() NOT NULL
+);
+--> statement-breakpoint
+ALTER TABLE "data_drain_runs" ADD CONSTRAINT "data_drain_runs_drain_id_data_drains_id_fk" FOREIGN KEY ("drain_id") REFERENCES "public"."data_drains"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint
+ALTER TABLE "data_drains" ADD CONSTRAINT "data_drains_organization_id_organization_id_fk" FOREIGN KEY ("organization_id") REFERENCES "public"."organization"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint
+ALTER TABLE "data_drains" ADD CONSTRAINT "data_drains_created_by_user_id_fk" FOREIGN KEY ("created_by") REFERENCES "public"."user"("id") ON DELETE no action ON UPDATE no action;--> statement-breakpoint
+CREATE INDEX "data_drain_runs_drain_started_idx" ON "data_drain_runs" USING btree ("drain_id","started_at");--> statement-breakpoint
+CREATE INDEX "data_drains_org_idx" ON "data_drains" USING btree ("organization_id");--> statement-breakpoint
+CREATE INDEX "data_drains_due_idx" ON "data_drains" USING btree ("enabled","last_run_at");--> statement-breakpoint
+CREATE UNIQUE INDEX "data_drains_org_name_unique" ON "data_drains" USING btree ("organization_id","name");--> statement-breakpoint
+CREATE INDEX "audit_log_workspace_created_at_id_idx" ON "audit_log" USING btree ("workspace_id",date_trunc('milliseconds', "created_at"),"id");--> statement-breakpoint
+CREATE INDEX "copilot_chats_workspace_created_at_id_idx" ON "copilot_chats" USING btree ("workspace_id",date_trunc('milliseconds', "created_at"),"id");--> statement-breakpoint
+CREATE INDEX "copilot_runs_workspace_completed_at_id_idx" ON "copilot_runs" USING btree ("workspace_id",date_trunc('milliseconds', "completed_at"),"id");--> statement-breakpoint
+CREATE INDEX "job_execution_logs_workspace_ended_at_id_idx" ON "job_execution_logs" USING btree ("workspace_id",date_trunc('milliseconds', "ended_at"),"id");--> statement-breakpoint
+CREATE INDEX "workflow_execution_logs_workspace_ended_at_id_idx" ON "workflow_execution_logs" USING btree ("workspace_id",date_trunc('milliseconds', "ended_at"),"id");

packages/db/migrations/meta/_journal.json

@@ -1422,6 +1422,13 @@
       "when": 1778022453620,
       "tag": "0203_curvy_quasimodo",
       "breakpoints": true
+    },
+    {
+      "idx": 204,
+      "version": "7",
+      "when": 1778024401275,
+      "tag": "0204_powerful_medusa",
+      "breakpoints": true
     }
   ]
 }

packages/db/schema.ts

@@ -354,6 +354,11 @@ export const workflowExecutionLogs = pgTable(
       table.workspaceId,
       table.startedAt
     ),
+    workspaceEndedAtIdIdx: index('workflow_execution_logs_workspace_ended_at_id_idx').on(
+      table.workspaceId,
+      sql`date_trunc('milliseconds', ${table.endedAt})`,
+      table.id
+    ),
     runningStartedAtIdx: index('workflow_execution_logs_running_started_at_idx')
       .on(table.startedAt)
       .where(sql`status = 'running'`),
@@ -588,6 +593,11 @@ export const jobExecutionLogs = pgTable(
       table.workspaceId,
       table.startedAt
     ),
+    workspaceEndedAtIdIdx: index('job_execution_logs_workspace_ended_at_id_idx').on(
+      table.workspaceId,
+      sql`date_trunc('milliseconds', ${table.endedAt})`,
+      table.id
+    ),
     executionIdUnique: uniqueIndex('job_execution_logs_execution_id_unique').on(table.executionId),
     triggerIdx: index('job_execution_logs_trigger_idx').on(table.trigger),
   })
@@ -1713,6 +1723,11 @@ export const copilotChats = pgTable(
     // Ordering indexes
     createdAtIdx: index('copilot_chats_created_at_idx').on(table.createdAt),
     updatedAtIdx: index('copilot_chats_updated_at_idx').on(table.updatedAt),
+    workspaceCreatedAtIdIdx: index('copilot_chats_workspace_created_at_id_idx').on(
+      table.workspaceId,
+      sql`date_trunc('milliseconds', ${table.createdAt})`,
+      table.id
+    ),
   })
 )
 
@@ -1844,6 +1859,11 @@ export const copilotRuns = pgTable(
       table.executionId,
       table.startedAt
     ),
+    workspaceCompletedAtIdIdx: index('copilot_runs_workspace_completed_at_id_idx').on(
+      table.workspaceId,
+      sql`date_trunc('milliseconds', ${table.completedAt})`,
+      table.id
+    ),
     streamIdUnique: uniqueIndex('copilot_runs_stream_id_unique').on(table.streamId),
   })
 )
@@ -2422,6 +2442,11 @@ export const auditLog = pgTable(
       table.workspaceId,
       table.createdAt
     ),
+    workspaceCreatedIdIdx: index('audit_log_workspace_created_at_id_idx').on(
+      table.workspaceId,
+      sql`date_trunc('milliseconds', ${table.createdAt})`,
+      table.id
+    ),
     actorCreatedIdx: index('audit_log_actor_created_idx').on(table.actorId, table.createdAt),
     resourceIdx: index('audit_log_resource_idx').on(table.resourceType, table.resourceId),
     actionIdx: index('audit_log_action_idx').on(table.action),
@@ -3094,3 +3119,90 @@ export const academyCertificate = pgTable(
     statusIdx: index('academy_certificate_status_idx').on(table.status),
   })
 )
+
+export const dataDrainSourceEnum = pgEnum('data_drain_source', [
+  'workflow_logs',
+  'job_logs',
+  'audit_logs',
+  'copilot_chats',
+  'copilot_runs',
+])
+
+export type DataDrainSource = (typeof dataDrainSourceEnum.enumValues)[number]
+
+export const dataDrainDestinationEnum = pgEnum('data_drain_destination', ['s3', 'webhook'])
+
+export type DataDrainDestination = (typeof dataDrainDestinationEnum.enumValues)[number]
+
+export const dataDrainCadenceEnum = pgEnum('data_drain_cadence', ['hourly', 'daily'])
+
+export type DataDrainCadence = (typeof dataDrainCadenceEnum.enumValues)[number]
+
+export const dataDrainRunStatusEnum = pgEnum('data_drain_run_status', [
+  'running',
+  'success',
+  'failed',
+])
+
+export type DataDrainRunStatus = (typeof dataDrainRunStatusEnum.enumValues)[number]
+
+export const dataDrainRunTriggerEnum = pgEnum('data_drain_run_trigger', ['cron', 'manual'])
+
+export type DataDrainRunTrigger = (typeof dataDrainRunTriggerEnum.enumValues)[number]
+
+export const dataDrains = pgTable(
+  'data_drains',
+  {
+    id: text('id').primaryKey(),
+    organizationId: text('organization_id')
+      .notNull()
+      .references(() => organization.id, { onDelete: 'cascade' }),
+    name: text('name').notNull(),
+    source: dataDrainSourceEnum('source').notNull(),
+    destinationType: dataDrainDestinationEnum('destination_type').notNull(),
+    /** Non-secret destination config (bucket, region, prefix, url, ...). Validated by destination registry. */
+    destinationConfig: jsonb('destination_config').$type<Record<string, unknown>>().notNull(),
+    /** Encrypted JSON blob containing destination credentials. Never returned to clients. */
+    destinationCredentials: text('destination_credentials').notNull(),
+    scheduleCadence: dataDrainCadenceEnum('schedule_cadence').notNull(),
+    enabled: boolean('enabled').notNull().default(true),
+    /** Opaque cursor — JSON-encoded, source-defined. Advances only on overall run success. */
+    cursor: text('cursor'),
+    lastRunAt: timestamp('last_run_at'),
+    lastSuccessAt: timestamp('last_success_at'),
+    createdBy: text('created_by')
+      .notNull()
+      .references(() => user.id),
+    createdAt: timestamp('created_at').notNull().defaultNow(),
+    updatedAt: timestamp('updated_at').notNull().defaultNow(),
+  },
+  (table) => ({
+    orgIdx: index('data_drains_org_idx').on(table.organizationId),
+    dueIdx: index('data_drains_due_idx').on(table.enabled, table.lastRunAt),
+    orgNameUnique: uniqueIndex('data_drains_org_name_unique').on(table.organizationId, table.name),
+  })
+)
+
+export const dataDrainRuns = pgTable(
+  'data_drain_runs',
+  {
+    id: text('id').primaryKey(),
+    drainId: text('drain_id')
+      .notNull()
+      .references(() => dataDrains.id, { onDelete: 'cascade' }),
+    status: dataDrainRunStatusEnum('status').notNull(),
+    trigger: dataDrainRunTriggerEnum('trigger').notNull(),
+    startedAt: timestamp('started_at').notNull().defaultNow(),
+    finishedAt: timestamp('finished_at'),
+    rowsExported: integer('rows_exported').notNull().default(0),
+    bytesWritten: bigint('bytes_written', { mode: 'number' }).notNull().default(0),
+    cursorBefore: text('cursor_before'),
+    cursorAfter: text('cursor_after'),
+    error: text('error'),
+    /** Destination-specific delivery locators for this run (e.g. S3 keys, webhook response ids). */
+    locators: jsonb('locators').$type<string[]>().notNull().default(sql`'[]'::jsonb`),
+  },
+  (table) => ({
+    drainStartedIdx: index('data_drain_runs_drain_started_idx').on(table.drainId, table.startedAt),
+  })
+)

packages/testing/src/mocks/audit.mock.ts

@@ -56,6 +56,11 @@ export const auditMock = {
     CUSTOM_TOOL_CREATED: 'custom_tool.created',
     CUSTOM_TOOL_UPDATED: 'custom_tool.updated',
     CUSTOM_TOOL_DELETED: 'custom_tool.deleted',
+    DATA_DRAIN_CREATED: 'data_drain.created',
+    DATA_DRAIN_UPDATED: 'data_drain.updated',
+    DATA_DRAIN_DELETED: 'data_drain.deleted',
+    DATA_DRAIN_RAN: 'data_drain.ran',
+    DATA_DRAIN_TESTED: 'data_drain.tested',
     CONNECTOR_DOCUMENT_RESTORED: 'connector_document.restored',
     CONNECTOR_DOCUMENT_EXCLUDED: 'connector_document.excluded',
     DOCUMENT_UPLOADED: 'document.uploaded',
@@ -156,6 +161,7 @@ export const auditMock = {
     CREDENTIAL: 'credential',
     CREDENTIAL_SET: 'credential_set',
     CUSTOM_TOOL: 'custom_tool',
+    DATA_DRAIN: 'data_drain',
     DOCUMENT: 'document',
     ENVIRONMENT: 'environment',
     FILE: 'file',

packages/testing/src/mocks/schema.mock.ts

@@ -1223,6 +1223,37 @@ export const schemaMock = {
     metadata: 'metadata',
     createdAt: 'createdAt',
   },
+  dataDrains: {
+    id: 'id',
+    organizationId: 'organizationId',
+    name: 'name',
+    source: 'source',
+    destinationType: 'destinationType',
+    destinationConfig: 'destinationConfig',
+    destinationCredentials: 'destinationCredentials',
+    scheduleCadence: 'scheduleCadence',
+    enabled: 'enabled',
+    cursor: 'cursor',
+    lastRunAt: 'lastRunAt',
+    lastSuccessAt: 'lastSuccessAt',
+    createdBy: 'createdBy',
+    createdAt: 'createdAt',
+    updatedAt: 'updatedAt',
+  },
+  dataDrainRuns: {
+    id: 'id',
+    drainId: 'drainId',
+    status: 'status',
+    trigger: 'trigger',
+    startedAt: 'startedAt',
+    finishedAt: 'finishedAt',
+    rowsExported: 'rowsExported',
+    bytesWritten: 'bytesWritten',
+    cursorBefore: 'cursorBefore',
+    cursorAfter: 'cursorAfter',
+    error: 'error',
+    locators: 'locators',
+  },
   /** Custom type export for tsvector */
   tsvector: 'tsvector',
 }