feat(dbmanager): implement enterprise dbmanager and integrate with co…

.gitignore

@@ -40,3 +40,6 @@ playwright-report/
 # Vendor Reference Code
 activepieces-reference/
 n8n-reference/
+
+# TypeScript Build Info
+*.tsbuildinfo

apps/api/package.json

@@ -41,6 +41,7 @@
     "@nexiom/cache": "workspace:*",
     "@nexiom/connections": "workspace:*",
     "@nexiom/database": "workspace:*",
+    "@nexiom/dbmanager": "workspace:*",
     "@nexiom/identity": "workspace:*",
     "bcryptjs": "^3.0.3",
     "better-auth": "^1.4.10",

apps/api/src/app/app.module.ts

@@ -21,6 +21,7 @@ import { TriggerModule } from '../modules/trigger/trigger.module';
 import { EmailModule } from '../modules/email/email.module';
 import { StorageResolverModule } from '../modules/storage-resolver/storage-resolver.module';
 import { CacheModule } from '@nexiom/cache';
+import { DbManagerModule } from '../modules/dbmanager/dbmanager.module';
 
 @Module({
   imports: [
@@ -82,6 +83,7 @@ import { CacheModule } from '@nexiom/cache';
     ConnectionsModule,
     TriggerModule,
     StorageResolverModule,
+    DbManagerModule,
   ],
   controllers: [AppController],
   providers: [AppService],

apps/api/src/modules/connections/connectors.service.spec.ts

@@ -7,6 +7,7 @@ import {
   EncryptionService,
   AppCredentialError,
 } from '@nexiom/connections';
+import { DB_MANAGER } from '../dbmanager/dbmanager.module';
 import {
   InternalServerErrorException,
   NotFoundException,
@@ -85,6 +86,7 @@ describe('ConnectorsService', () => {
         { provide: ConfigService, useValue: mockConfigService },
         { provide: EncryptionService, useValue: mockEncryptionService },
         { provide: DATABASE_CONNECTION, useValue: mockDb as unknown },
+        { provide: DB_MANAGER, useValue: { applyPlan: vi.fn() } },
       ],
     }).compile();
 
@@ -561,7 +563,7 @@ describe('ConnectorsService', () => {
       ).rejects.toThrow(InternalServerErrorException);
     });
 
-    it('should throw InternalServerErrorException and abort if schema creation execute() fails', async () => {
+    it('should throw InternalServerErrorException and abort if DB manager applyPlan fails', async () => {
       // Mock appConnection insert succeeding:
       mockDbInsert.mockReturnValueOnce({
         values: vi.fn().mockReturnValue({
@@ -582,11 +584,29 @@ describe('ConnectorsService', () => {
         }),
       });
 
-      // Mock the raw schema execute command failing:
-      mockDb.execute.mockRejectedValueOnce(new Error('schema create failed'));
+      // We need to re-mock the DB_MANAGER for this specific test
+      const failingDbManager = {
+        applyPlan: vi.fn().mockRejectedValue(new Error('applyPlan failed')),
+      };
+      const moduleFail: TestingModule = await Test.createTestingModule({
+        providers: [
+          ConnectorsService,
+          { provide: ProviderRegistryService, useValue: mockProviderRegistry },
+          {
+            provide: ConfigService,
+            useValue: { get: vi.fn().mockReturnValue('mock-region-context') },
+          },
+          { provide: EncryptionService, useValue: mockEncryptionService },
+          { provide: DATABASE_CONNECTION, useValue: mockDb as unknown },
+          { provide: DB_MANAGER, useValue: failingDbManager },
+        ],
+      }).compile();
+
+      const failingService =
+        moduleFail.get<ConnectorsService>(ConnectorsService);
 
       await expect(
-        service.storeOAuthConnection({
+        failingService.storeOAuthConnection({
           tenantId: 'tenant-123',
           providerName: 'salesforce',
           externalId: 'salesforce-tms',

apps/api/src/modules/connections/connectors.service.ts

@@ -20,8 +20,9 @@ import {
   DATABASE_CONNECTION,
   type DrizzleDb,
 } from '@nexiom/database';
-import { sql } from 'drizzle-orm';
-import * as crypto from 'crypto';
+import { DatabaseManager, SchemaPlan } from '@nexiom/dbmanager';
+import { DB_MANAGER } from '../dbmanager/dbmanager.module';
+import * as crypto from 'node:crypto';
 
 /** Encrypted value blob stored in app_connection.value — mirrors Activepieces BaseOAuth2ConnectionValue */
 export interface ConnectionValueBlob {
@@ -55,6 +56,7 @@ export class ConnectorsService {
 
   constructor(
     @Inject(DATABASE_CONNECTION) private readonly db: DrizzleDb,
+    @Inject(DB_MANAGER) private readonly dbManager: DatabaseManager,
     private readonly providerRegistry: ProviderRegistryService,
     private readonly configService: ConfigService,
   ) {}
@@ -297,7 +299,7 @@ export class ConnectorsService {
     }
 
     try {
-      await this.db.transaction(async (tx) => {
+      const workspaceSchemaName = await this.db.transaction(async (tx) => {
         // 1. Insert or update the business connection metadata
         const [connection] = await tx
           .insert(appConnections)
@@ -340,27 +342,30 @@ export class ConnectorsService {
         const sanitizedProvider = providerName.replaceAll(/[^a-z0-9]/g, '');
         const finalProviderToken = sanitizedProvider || 'unknown';
         const safeToken = finalProviderToken.substring(0, 40);
-        const workspaceSchemaName = `ws_${safeToken}_${hashedSuffix}`;
+        const schemaName = `ws_${safeToken}_${hashedSuffix}`;
 
         await tx
           .insert(connectionStorageRegistry)
           .values({
             connectionId: connection.id,
-            workspaceId: workspaceSchemaName,
+            workspaceId: schemaName,
             databaseHostId: 'primary-cluster', // Can be parameterized later for regional sharding
             regionContext: finalRegionContext,
           })
           .onConflictDoNothing({
             target: connectionStorageRegistry.connectionId,
           }); // Already provisioned
 
-        // 3. Actually create the physical PostgreSQL schema on the cluster
-        // Using sql.raw here is safe because workspaceSchemaName is strictly internally generated
-        // from a regex-sanitized string and a UUID slice, not user input.
-        await tx.execute(
-          sql`CREATE SCHEMA IF NOT EXISTS "${sql.raw(workspaceSchemaName)}"`,
-        );
+        return schemaName;
       });
+
+      // 3. Apply the initial schema plan (NAMESPACE_ONLY) outside the transaction
+      // so the DDL runs on its own connection and does not silently escape the
+      // Drizzle tx scope (db.$client vs the transaction's dedicated connection).
+      await this.dbManager.applyPlan(
+        workspaceSchemaName,
+        SchemaPlan.NAMESPACE_ONLY,
+      );
     } catch (error) {
       if (error instanceof HttpException) {
         throw error;

apps/api/src/modules/dbmanager/dbmanager.module.ts

@@ -0,0 +1,23 @@
+import { Module, Global } from '@nestjs/common';
+import { SqlDatabaseManager } from '@nexiom/dbmanager';
+import { DbModule } from '../../db/db.module';
+import { DATABASE_CONNECTION } from '@nexiom/database';
+import type { DrizzleDb } from '@nexiom/database';
+
+export const DB_MANAGER = 'DATABASE_MANAGER';
+
+@Global()
+@Module({
+  imports: [DbModule],
+  providers: [
+    {
+      provide: DB_MANAGER,
+      useFactory: (drizzleDb: DrizzleDb) => {
+        return new SqlDatabaseManager(drizzleDb);
+      },
+      inject: [DATABASE_CONNECTION],
+    },
+  ],
+  exports: [DB_MANAGER],
+})
+export class DbManagerModule {}

apps/api/src/modules/trigger/trigger-executor.service.spec.ts

@@ -46,6 +46,9 @@ describe('TriggerExecutorService', () => {
     service = new TriggerExecutorService(
       db as unknown as import('@nexiom/database').DrizzleDb,
       redis as unknown as import('ioredis').Redis,
+      {
+        applyPlan: vi.fn(),
+      } as unknown as import('@nexiom/dbmanager').DatabaseManager,
     );
   });
 

apps/api/src/modules/trigger/trigger-executor.service.ts

@@ -2,6 +2,8 @@ import { Injectable, Inject, Logger } from '@nestjs/common';
 import type { Trigger, TriggerContext } from '@nexiom/connections';
 import type { DrizzleDb } from '@nexiom/database';
 import { DATABASE_CONNECTION } from '@nexiom/database';
+import { DatabaseManager, SchemaPlan } from '@nexiom/dbmanager';
+import { DB_MANAGER } from '../dbmanager/dbmanager.module';
 import type { Redis } from 'ioredis';
 import { createHash, randomUUID } from 'node:crypto';
 import { RedisBackedTriggerStore } from './redis-trigger-store';
@@ -59,6 +61,7 @@ export class TriggerExecutorService {
   constructor(
     @Inject(DATABASE_CONNECTION) private readonly db: DrizzleDb,
     @Inject('REDIS_CLIENT') private readonly redis: Redis,
+    @Inject(DB_MANAGER) private readonly dbManager: DatabaseManager,
   ) {}
 
   // ─── Polling ────────────────────────────────────────────────────────────
@@ -138,6 +141,13 @@ export class TriggerExecutorService {
   async runOnEnable(params: TriggerRunParams): Promise<void> {
     const context = this.buildContext(params);
     try {
+      // 1. Ensure the gateway tables are provisioned lazily
+      await this.dbManager.applyPlan(
+        params.workspaceId,
+        SchemaPlan.GATEWAY_ACTIVE,
+      );
+
+      // 2. Invoke the trigger enablement logic (e.g. Subscribe to webhook)
       await params.trigger.onEnable?.(context);
       this.logger.log('onEnable completed', {
         appName: params.appName,
@@ -428,7 +438,7 @@ export class TriggerExecutorService {
     ) {
       const sorted = Object.keys(record as Record<string, unknown>)
         .filter((k) => !VOLATILE_KEYS.has(k))
-        .sort((a, b) => (a < b ? -1 : a > b ? 1 : 0))
+        .sort((a, b) => (a ?? '').localeCompare(b ?? ''))
         .reduce<Record<string, unknown>>((acc, k) => {
           acc[k] = (record as Record<string, unknown>)[k];
           return acc;

apps/api/src/modules/trigger/trigger.module.ts

@@ -9,6 +9,8 @@ import { DlqProcessorService } from './dlq-processor.service';
 import { WebhooksController } from './webhooks.controller';
 import { DATABASE_CONNECTION } from '@nexiom/database';
 import type { DrizzleDb } from '@nexiom/database';
+import { DB_MANAGER } from '../dbmanager/dbmanager.module';
+import { DatabaseManager } from '@nexiom/dbmanager';
 
 /**
  * Wires all trigger-related services.
@@ -23,9 +25,9 @@ import type { DrizzleDb } from '@nexiom/database';
     PieceRegistryService,
     {
       provide: TriggerExecutorService,
-      useFactory: (db: DrizzleDb, redis: Redis) =>
-        new TriggerExecutorService(db, redis),
-      inject: [DATABASE_CONNECTION, REDIS_CLIENT],
+      useFactory: (db: DrizzleDb, redis: Redis, dbManager: DatabaseManager) =>
+        new TriggerExecutorService(db, redis, dbManager),
+      inject: [DATABASE_CONNECTION, REDIS_CLIENT, DB_MANAGER],
     },
     {
       provide: PollerService,

docs/architecture/schema_lifecycle_and_provisioning.md

@@ -0,0 +1,59 @@
+# Architecture: Schema Lifecycle & Lazy Provisioning
+
+This document defines the timing and triggers for database schema creation within the FluxNex 6-layer pipeline. To ensure maximum scalability, we follow a Lazy Provisioning strategy—nothing is created until there is active intent from the user.
+
+1. Provisioning Lifecycle Summary
+| Lifecycle Event | Physical Action | Component Responsible | Status |
+| --------------- | --------------- | ----------------------- | ------ |
+| App Authentication | Create Schema Namespace (ws_{id}) | api-gateway / dbmanager | Empty Silo |
+| Route Activation | Create Kernel Tables (Ingest/Logs) | dbmanager (Atlas) | Active Silo |
+| Object Selection | Initialize Unified Replica (replica_entity) | dbmanager (Atlas) | Data Ready |
+
+2. Event-Driven Provisioning Details
+Stage 1: Connection Handshake (The "Namespace" Event)
+Trigger: User successfully completes an OAuth or API Key handshake.
+Database Action: CREATE SCHEMA IF NOT EXISTS "ws_sf_101".
+Registry Action: Insert row into public.connection_storage_registry.
+Result: A physical "Wall" is established. The schema is empty, incurring zero index or metadata overhead in the Postgres catalog.
+Stage 2: Route Activation (The "Kernel" Event)
+Trigger: User creates a Route (e.g., Salesforce → QuickBooks) and clicks "Save & Activate."
+Database Action: The dbmanager uses Atlas to deploy the Pipeline Kernel tables into the connection's schema.
+Tables Created: * `inbound_gateway`: Stores raw ingestion data.
+*(Note: `sync_log`, `sync_cursor`, and `outbound_gateway` are planned additions for future Gateway integration).*
+Stage 3: Object Mapping (The "Storage Initialization" Event)
+Trigger: User selects a specific object (e.g., "Accounts") in the Mapping UI.
+Database Action: The system ensures the structured state store is ready. It executes the SQL to create:
+replica_entity: The unified JSONB state store for ALL objects in this connection.
+GIN Indexes: Applied to the data column to enable high-speed search across all fields and custom objects.
+Optimization: This table is created once per connection and handles all object types (Account, Contact, etc.) through the entity_type column.
+
+3. The 6-Layer Readiness Matrix
+This matrix shows which physical tables must be "ready" before a specific layer worker can process a job.
+| Layer | Dependency | Physical Location | Why? |
+| ----- | ---------- | ----------------- | ---- |
+| L1: Source Gateway | inbound_gateway | ws_source | To log the "evidence" of the raw webhook. |
+| L2: Universal Replica | replica_entity | ws_source | To store the parsed source-of-truth. |
+| L3: Normalization | normalized_entity | ws_source | To store the platform-standard JSON. |
+| L4: Outbound Prep | field_mapping | public | To read the rules defined by the user. |
+| L5: Delivery Engine | N/A | N/A | Pure I/O. Does not write to tenant schemas. |
+| L6: Dest. Gateway | outbound_gateway | ws_dest | To log the destination API response. |
+
+4. Engineering Implementation: The "Ensure" Pattern
+Because workers are asynchronous, a webhook (L1) might arrive before the dbmanager has finished creating the tables. To prevent crashes, all workers follow the Ensure Pattern:
+// packages/engine/src/workers/base.worker.ts
+async function processLayerJob(connectionId: string, payload: any) {
+  // 1. Resolve Schema Name
+  const schema = await storageResolver.resolve(connectionId);
+
+  // 2. JIT Check (The "Ensure" step)
+  // If the tables are missing, the worker triggers a fast schema plan apply
+  await dbmanager.applyPlan(schema, SchemaPlan.GATEWAY_ACTIVE);
+
+  // 3. Perform Business Logic
+  await db.withSchema(schema).insert(...);
+}
+
+1. Benefits of this Timing
+Catalog Leanliness: We avoid the "Postgres Metadata Bloat" that occurs when thousands of unused tables exist in the system.
+Speed: App connection is instantaneous because it only creates a namespace.
+Isolation: The architecture guarantees that no worker ever attempts to write to a "Shared" schema; it is always scoped to the connection-specific workspace.