Tenants: Bulk Sync Connection Paths (JavaScript)

// This script will sync a specified path for all tenant connections, using pagination.
// It finds all tenants, gets their connections, and triggers a sync for the specified path.
// 1: Load environment variables from a specific .env file for configuration
require('dotenv').config({ path: 'sigma-api-recipes/.env' });
const crypto = require('crypto');
const axios = require('axios');
// 2: Load variables from environment
const baseURL = process.env.baseURL;
const clientId = process.env.CLIENT_ID;
const clientSecret = process.env.CLIENT_SECRET;
// 3: Configuration
const CONNECTION_NAME_FILTER = 'My Connection'; // Optional: filter by connection name (set to null to sync all connections)
const CONCURRENCY = 3; // Number of parallel requests
// Note that this can be used to sync any path, not just a table.
// Different data platforms have different path length requirements (e.g., Redshift uses 2 levels, while Snowflake uses 3).
const SYNC_PATH = ['MY_DATABASE', 'MY_SCHEMA', 'MY_TABLE']; // Path to sync (database, schema, table)
// Optional: Token cache to avoid unnecessary token requests
let parentTokenCache = { token: '', expiresAt: 0 };
const tenantTokenCache = new Map();
function base64UrlEncode(data) {
    return Buffer.from(data)
        .toString('base64')
        .replace(/\+/g, '-')
        .replace(/\//g, '_')
        .replace(/=+$/, '');
}
// Get parent org access token
async function getParentToken() {
    if (parentTokenCache.token && Date.now() < parentTokenCache.expiresAt - 60000) {
        return parentTokenCache.token;
    }
    const params = new URLSearchParams({
        grant_type: 'client_credentials',
        client_id: clientId,
        client_secret: clientSecret,
    });
    const response = await axios.post(`${baseURL}/v2/auth/token`, params.toString(), {
        headers: { 'Content-Type': 'application/x-www-form-urlencoded' }
    });
    parentTokenCache = {
        token: response.data.access_token,
        expiresAt: Date.now() + response.data.expires_in * 1000
    };
    return parentTokenCache.token;
}
// Create self-signed JWT subject token for tenant impersonation
function createSubjectToken(tenantOrganizationId) {
    const now = Math.floor(Date.now() / 1000);
    const header = { alg: 'HS256', typ: 'JWT', kid: clientId };
    const payload = { iat: now, exp: now + 60, tenant: tenantOrganizationId };
    const encodedHeader = base64UrlEncode(JSON.stringify(header));
    const encodedPayload = base64UrlEncode(JSON.stringify(payload));
    const signatureInput = `${encodedHeader}.${encodedPayload}`;
    const signature = crypto.createHmac('sha256', clientSecret)
        .update(signatureInput)
        .digest('base64')
        .replace(/\+/g, '-')
        .replace(/\//g, '_')
        .replace(/=+$/, '');
    return `${signatureInput}.${signature}`;
}
// Exchange parent token for tenant-scoped token
async function getTenantToken(tenantOrganizationId) {
    const cached = tenantTokenCache.get(tenantOrganizationId);
    if (cached && Date.now() < cached.expiresAt - 60000) {
        return cached.token;
    }
    const actorToken = await getParentToken();
    const subjectToken = createSubjectToken(tenantOrganizationId);
    const params = new URLSearchParams({
        grant_type: 'urn:ietf:params:oauth:grant-type:token-exchange',
        subject_token: subjectToken,
        subject_token_type: 'urn:ietf:params:oauth:token-type:jwt',
        actor_token: actorToken,
        actor_token_type: 'urn:ietf:params:oauth:token-type:access_token',
    });
    const response = await axios.post(`${baseURL}/v2/auth/token`, params.toString(), {
        headers: {
            'Content-Type': 'application/x-www-form-urlencoded',
            Authorization: `Bearer ${actorToken}`
        }
    });
    tenantTokenCache.set(tenantOrganizationId, {
        token: response.data.access_token,
        expiresAt: Date.now() + response.data.expires_in * 1000
    });
    return response.data.access_token;
}
// Get all tenants using pagination
async function getAllTenants() {
    const token = await getParentToken();
    let nextPageToken = null;
    const allTenants = [];
    do {
        const url = nextPageToken
            ? `${baseURL}/v2/tenants?pageToken=${nextPageToken}`
            : `${baseURL}/v2/tenants`;
        const response = await axios.get(url, {
            headers: { Authorization: `Bearer ${token}` }
        });
        allTenants.push(...response.data.entries.filter(e => e !== null));
        nextPageToken = response.data.nextPageToken || null;
    } while (nextPageToken);
    console.log(`Found ${allTenants.length} tenants`);
    return allTenants;
}
// Get connections for a tenant (using tenant-scoped auth token)
async function getTenantConnections(tenantOrganizationId) {
    const token = await getTenantToken(tenantOrganizationId);
    const allConnections = [];
    let nextPage = null;
    do {
        const params = new URLSearchParams({ limit: '100' });
        if (nextPage) params.set('page', nextPage);
        const response = await axios.get(`${baseURL}/v2/connections?${params}`, {
            headers: { Authorization: `Bearer ${token}` }
        });
        allConnections.push(...response.data.entries);
        nextPage = response.data.nextPage || null;
    } while (nextPage);
    return allConnections;
}
// Sync a connection by path (using tenant-scoped token)
async function syncConnection(tenantOrganizationId, connectionId, path) {
    const token = await getTenantToken(tenantOrganizationId);
    const startTime = Date.now();
    await axios.post(
        `${baseURL}/v2/connections/${connectionId}/sync`,
        { path },
        { headers: { Authorization: `Bearer ${token}`, 'Content-Type': 'application/json' } }
    );
    return { latencyMs: Date.now() - startTime };
}
// Function to process items with controlled concurrency
async function processWithConcurrency(items, concurrency, processor) {
    const results = [];
    const executing = new Set();
    for (const item of items) {
        const promise = processor(item).then(result => {
            executing.delete(promise);
            return result;
        });
        executing.add(promise);
        results.push(promise);
        if (executing.size >= concurrency) {
            await Promise.race(executing);
        }
    }
    return Promise.all(results);
}
// Main function to sync connection paths across all tenants
async function syncTenantConnectionPaths() {
    const startTime = Date.now();
    // Step 1: Get all tenants
    console.log('\n=== Fetching Tenants ===');
    const tenants = await getAllTenants();
    // Step 2: Process each tenant
    console.log('\n=== Syncing Connection Paths ===');
    const results = [];
    await processWithConcurrency(tenants, CONCURRENCY, async (tenant) => {
        try {
            // Get connections for this tenant
            const connections = await getTenantConnections(tenant.organizationId);
            // Filter by connection name
            // Note: similar APIs could be used to retrieve connections via the source swap policy
            const targetConnections = CONNECTION_NAME_FILTER
                ? connections.filter(c => c.name === CONNECTION_NAME_FILTER)
                : connections;
            if (targetConnections.length === 0) {
                results.push({
                    tenantName: tenant.name,
                    status: 'skipped',
                    reason: CONNECTION_NAME_FILTER
                        ? `No connection named "${CONNECTION_NAME_FILTER}"`
                        : 'No connections found'
                });
                return;
            }
            // Sync each matching connection
            for (const connection of targetConnections) {
                try {
                    const syncResult = await syncConnection(
                        tenant.organizationId,
                        connection.connectionId,
                        SYNC_PATH
                    );
                    results.push({
                        tenantName: tenant.name,
                        connectionName: connection.name,
                        status: 'success',
                        latencyMs: syncResult.latencyMs
                    });
                    console.log(`✓ ${tenant.name} - ${connection.name} synced in ${syncResult.latencyMs}ms`);
                } catch (error) {
                    results.push({
                        tenantName: tenant.name,
                        connectionName: connection.name,
                        status: 'failed',
                        error: error.response?.data?.message || error.message
                    });
                    console.error(`✗ ${tenant.name} - ${connection.name}: ${error.message}`);
                }
            }
        } catch (error) {
            results.push({
                tenantName: tenant.name,
                status: 'failed',
                error: error.response?.data?.message || error.message
            });
            console.error(`✗ ${tenant.name}: ${error.message}`);
        }
    });
    // Step 3: Print summary
    const totalElapsedMs = Date.now() - startTime;
    const successes = results.filter(r => r.status === 'success');
    const failures = results.filter(r => r.status === 'failed');
    const skipped = results.filter(r => r.status === 'skipped');
    console.log('\n=== Summary ===');
    console.log(`Path synced: ${SYNC_PATH.join('.')}`);
    console.log(`Total tenants: ${tenants.length}`);
    console.log(`Successful syncs: ${successes.length}`);
    console.log(`Failed syncs: ${failures.length}`);
    console.log(`Skipped: ${skipped.length}`);
    console.log(`Total elapsed time: ${(totalElapsedMs / 1000).toFixed(2)}s`);
    if (successes.length > 0) {
        const latencies = successes.map(r => r.latencyMs).sort((a, b) => a - b);
        const avgLatency = latencies.reduce((a, b) => a + b, 0) / latencies.length;
        console.log(`\nLatency stats:`);
        console.log(`  Min: ${latencies[0]}ms`);
        console.log(`  Max: ${latencies[latencies.length - 1]}ms`);
        console.log(`  Avg: ${avgLatency.toFixed(0)}ms`);
    }
    if (failures.length > 0) {
        console.log('\nFailed syncs:');
        console.table(failures.slice(0, 10).map(f => ({
            Tenant: f.tenantName,
            Connection: f.connectionName || 'N/A',
            Error: f.error
        })));
    }
}
// Execute the main function
syncTenantConnectionPaths().catch(error => {
    console.error('Failed to sync connection paths:', error);
});

✓ Tenant A - My Connection synced in 245ms
✓ Tenant B - My Connection synced in 198ms
✓ Tenant C - My Connection synced in 312ms
=== Summary ===
Path synced: MY_DATABASE.MY_SCHEMA.MY_TABLE
Total tenants: 3
Successful syncs: 3
Failed syncs: 0
Skipped: 0
Total elapsed time: 1.23s
Latency stats:
  Min: 198ms
  Max: 312ms
  Avg: 252ms