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

[waleedlatif1]

Summary

• Continuously exports workflow logs, job logs, audit logs, copilot chats, and copilot runs to customer-owned S3 buckets or HTTPS webhooks on hourly or daily schedules
• Pairs with data retention so customers can drain into long-term storage before Sim deletes
• Built on two registries (DrainSource + DrainDestination) so future destinations are a single-file change
• At-least-once delivery via opaque cursor that advances only on full success; consumers dedupe on stable row ids
• SSRF-validated webhooks with DNS pinning, HMAC-SHA256 timestamp signatures, S3 server-side encryption, audit logging on every config and run change
• Self-hosted gating via DATA_DRAINS_ENABLED / NEXT_PUBLIC_DATA_DRAINS_ENABLED, mirroring data retention

Type of Change

• New feature

Testing

• 26 unit tests passing (service, dispatcher, sources, S3, webhook)
• bun run check:api-validation passing
• Manually tested S3 + webhook end-to-end including failure paths and cursor replay

Checklist

• Code follows project style guidelines
• Self-reviewed my changes
• Tests added/updated and passing
• No new warnings introduced
• I confirm that I have read and agree to the terms outlined in the Contributor License Agreement (CLA)

[vercel]

The latest updates on your projects. Learn more about Vercel for GitHub.

Project	Deployment	Actions	Updated (UTC)
docs	Ready	Preview, Comment	May 5, 2026 5:50pm

[cursor]

PR Summary

High Risk
Introduces new org-scoped export infrastructure (new tables, API routes, background jobs, and outbound network writes to S3/webhooks), which is security- and data-handling sensitive. Risk centers on authz gating, SSRF protections, and correct cursor/run-state handling under failures/retries.

Overview
Adds a new Enterprise “Data Drains” feature to continuously export workflow/job/audit/copilot data as NDJSON to either S3 or an HTTPS webhook, including a Settings UI to create/manage drains, run them manually, test destinations, and view recent run history.

Implements the backend: new data_drains/data_drain_runs schema + migration, org/admin-only API endpoints (create/list/get/update/delete, run now, test, list runs) with audit logging, a cron dispatcher (/api/cron/run-data-drains) that enqueues run-data-drain jobs, and a runner service that does chunked export with at-least-once semantics (cursor advances only on full success) and orphaned-run reaping.

Adds destination/source registries with initial implementations for S3 (deterministic keys, SSE AES256, optional endpoint SSRF validation) and webhook (DNS pinning + SSRF validation, HMAC-signed requests, retries/backoff, idempotency headers), plus new self-hosted feature flags (DATA_DRAINS_ENABLED / NEXT_PUBLIC_DATA_DRAINS_ENABLED) and Helm cron/env wiring, and extends secureFetch to support AbortSignal.

^{Reviewed by Cursor Bugbot for commit 57f6571. Configure here.}

[greptile-apps]

Greptile Summary

This PR introduces a full data-drain pipeline for continuously exporting workflow logs, job logs, audit logs, copilot chats, and copilot runs to customer-owned S3 buckets or HTTPS webhooks on hourly/daily schedules. It includes two destination backends (S3 + webhook), five source generators with composite keyset cursors, a dispatcher with conditional atomic claiming, an orphan reaper, Trigger.dev background task integration, and SSRF-safe delivery with HMAC-SHA256 signatures.

• Dispatcher & service: At-least-once delivery via cursor advancing only on full success; cancellation signal is respected post-loop and marks the run as failed; per-org enterprise checks are wrapped in try-catch to prevent one billing-API failure from aborting the entire batch.
• Security: Webhook destination uses validateUrlWithDNS + IP pinning (secureFetchWithPinnedIP) to defeat DNS rebinding; S3 endpoint is SSRF-validated at both the schema (sync IP-literal check) and session-open (DNS check) layers; credentials are encrypted with AES-256-GCM before storage.
• Access control: Feature-flag and enterprise-plan gates apply to both reads and writes; owner/admin role is required for all drain operations.

Confidence Score: 5/5

Safe to merge; the core delivery guarantees, security hardening, and access control gates are all correct.

This is a large, well-structured feature with thorough security controls. All findings are minor: one metrics counter undercount in the dispatcher, one hard-parse failure mode that could affect list responsiveness after a schema evolution, and a missing Trigger.dev maxDuration that could impede very large first-time backfills. None of these affect data integrity, security, or correctness for the typical enterprise drain workflow.

apps/sim/lib/data-drains/dispatcher.ts (skipped counter), apps/sim/lib/data-drains/serializers.ts (hard-parse list failure), apps/sim/background/run-data-drain.ts (no maxDuration)

Important Files Changed

Filename	Overview
apps/sim/lib/data-drains/dispatcher.ts	Dispatch loop logic is solid with conditional claim, concurrency key, and per-org enterprise check in try-catch. Minor gap: failed-enqueue drains increment neither dispatched nor skipped, so candidates − dispatched − skipped can be non-zero without explanation.
apps/sim/lib/data-drains/service.ts	Correctly orchestrates drain runs: at-least-once delivery via cursor only advancing on full success, signal.aborted check post-loop marking cancellations as failed, and session always closed in finally.
apps/sim/lib/data-drains/destinations/webhook.ts	SSRF-validated with DNS pinning, HMAC-SHA256 Stripe-style signatures, signal threaded into fetch, exponential backoff with jitter, and Retry-After parsing.
apps/sim/lib/data-drains/destinations/s3.ts	Endpoint SSRF-validated at both schema level (sync IP-literal check) and runtime (DNS check via assertEndpointIsPublic). SSE-AES256 on all writes. Lazy cached endpoint check amortizes DNS across chunks.
apps/sim/lib/data-drains/access.ts	Feature-flag and enterprise-plan gates now apply to both reads and writes. Role check (owner/admin) is consistent across all endpoints.
apps/sim/lib/data-drains/serializers.ts	Re-validates stored JSONB config via configSchema.parse() before serialization. Using .parse() (throws) rather than .safeParse() means one drain with a schema-invalid config will throw from the list endpoint for all drains in that org.
apps/sim/lib/data-drains/sources/cursor.ts	Composite (timestamp, id) keyset pagination correctly handles ties; returns undefined for null cursor so Drizzle's and() skips the predicate on first run.
apps/sim/background/run-data-drain.ts	Thin Trigger.dev task wrapper correctly threads the cancellation signal into runDrain. No maxDuration configured — very large backfills that exceed the default Trigger.dev timeout will time out, mark as failed, and restart from cursorBefore indefinitely.
apps/sim/lib/data-drains/encryption.ts	Thin wrapper around shared AES-256-GCM helper. Callers are expected to run credentialsSchema.parse() after decryption, which both the service and test route do correctly.
apps/sim/lib/api/contracts/data-drains.ts	API contracts correctly separate public endpoint schema (no SSRF check) from the destination-specific runtime schema. Discriminated union for create enforces destinationType; loose record schema for update is intentional and re-validated at the route layer.
packages/db/schema.ts	New dataDrains and dataDrainRuns tables look correct. rowsExported/bytesWritten have DB-level NOT NULL DEFAULT 0, locators defaults to []::jsonb, cascade delete from dataDrains to dataDrainRuns.

Sequence Diagram

sequenceDiagram
    participant Cron as Cron (/run-data-drains)
    participant Dispatcher
    participant DB
    participant Queue as Job Queue (Trigger.dev)
    participant Task as run-data-drain task
    participant Source
    participant Destination

    Cron->>Dispatcher: dispatchDueDrains()
    Dispatcher->>DB: reapOrphanedRuns()
    Dispatcher->>DB: SELECT due drains
    loop per candidate
        Dispatcher->>DB: isOrganizationOnEnterprisePlan
        Dispatcher->>DB: UPDATE claim (conditional)
        Dispatcher->>Queue: enqueue run-data-drain
    end
    Queue->>Task: trigger with signal
    Task->>DB: SELECT drain + config
    Task->>Task: decryptCredentials
    Task->>DB: INSERT dataDrainRun status=running
    Task->>Destination: openSession()
    loop each chunk
        Task->>Source: pages() chunk[]
        Task->>Destination: deliver(body, metadata, signal)
        Destination-->>Task: locator
    end
    alt success
        Task->>DB: UPDATE dataDrains cursor + lastSuccessAt
        Task->>DB: UPDATE dataDrainRun status=success
    else cancelled / error
        Task->>DB: UPDATE dataDrains lastRunAt only
        Task->>DB: UPDATE dataDrainRun status=failed
    end
    Task->>Destination: close()

Loading

_{Reviews (12): Last reviewed commit: "refactor(data-drains): trim extraneous c..." | Re-trigger Greptile}

[waleedlatif1]

@greptile

[waleedlatif1]

@cursor review

[waleedlatif1]

@greptile

[waleedlatif1]

@cursor review

[waleedlatif1]

@greptile

[waleedlatif1]

@cursor review

[waleedlatif1]

@greptile

[waleedlatif1]

@cursor review

[waleedlatif1]

@greptile

[waleedlatif1]

@cursor review

[waleedlatif1]

@greptile

[waleedlatif1]

@greptile

[waleedlatif1]

@cursor review

[cursor]

✅ Bugbot reviewed your changes and found no new issues!

Comment @cursor review or bugbot run to trigger another review on this PR

^{Reviewed by Cursor Bugbot for commit 4fe0d0e. Configure here.}

[waleedlatif1]

@greptile

[waleedlatif1]

@cursor review

[cursor]

Cursor Bugbot has reviewed your changes and found 3 potential issues.

^{❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.}

^{Reviewed by Cursor Bugbot for commit 57f6571. Configure here.}