Tenants: Bulk Sync Connection Paths (JavaScript)

1
// This script will sync a specified path for all tenant connections, using pagination.
2
// It finds all tenants, gets their connections, and triggers a sync for the specified path.
3
4
// 1: Load environment variables from a specific .env file for configuration
5
require('dotenv').config({ path: 'sigma-api-recipes/.env' });
6
const crypto = require('crypto');
7
const axios = require('axios');
8
9
// 2: Load variables from environment
10
const baseURL = process.env.baseURL;
11
const clientId = process.env.CLIENT_ID;
12
const clientSecret = process.env.CLIENT_SECRET;
13
14
// 3: Configuration
15
const CONNECTION_NAME_FILTER = 'My Connection'; // Optional: filter by connection name (set to null to sync all connections)
16
const CONCURRENCY = 3; // Number of parallel requests
17
18
// Note that this can be used to sync any path, not just a table.
19
// Different data platforms have different path length requirements (e.g., Redshift uses 2 levels, while Snowflake uses 3).
20
const SYNC_PATH = ['MY_DATABASE', 'MY_SCHEMA', 'MY_TABLE']; // Path to sync (database, schema, table)
21
22
// Optional: Token cache to avoid unnecessary token requests
23
let parentTokenCache = { token: '', expiresAt: 0 };
24
const tenantTokenCache = new Map();
25
26
function base64UrlEncode(data) {
27
    return Buffer.from(data)
28
        .toString('base64')
29
        .replace(/\+/g, '-')
30
        .replace(/\//g, '_')
31
        .replace(/=+$/, '');
32
}
33
34
// Get parent org access token
35
async function getParentToken() {
36
    if (parentTokenCache.token && Date.now() < parentTokenCache.expiresAt - 60000) {
37
        return parentTokenCache.token;
38
    }
39
40
    const params = new URLSearchParams({
41
        grant_type: 'client_credentials',
42
        client_id: clientId,
43
        client_secret: clientSecret,
44
    });
45
46
    const response = await axios.post(`${baseURL}/v2/auth/token`, params.toString(), {
47
        headers: { 'Content-Type': 'application/x-www-form-urlencoded' }
48
    });
49
50
    parentTokenCache = {
51
        token: response.data.access_token,
52
        expiresAt: Date.now() + response.data.expires_in * 1000
53
    };
54
55
    return parentTokenCache.token;
56
}
57
58
// Create self-signed JWT subject token for tenant impersonation
59
function createSubjectToken(tenantOrganizationId) {
60
    const now = Math.floor(Date.now() / 1000);
61
62
    const header = { alg: 'HS256', typ: 'JWT', kid: clientId };
63
    const payload = { iat: now, exp: now + 60, tenant: tenantOrganizationId };
64
65
    const encodedHeader = base64UrlEncode(JSON.stringify(header));
66
    const encodedPayload = base64UrlEncode(JSON.stringify(payload));
67
    const signatureInput = `${encodedHeader}.${encodedPayload}`;
68
69
    const signature = crypto.createHmac('sha256', clientSecret)
70
        .update(signatureInput)
71
        .digest('base64')
72
        .replace(/\+/g, '-')
73
        .replace(/\//g, '_')
74
        .replace(/=+$/, '');
75
76
    return `${signatureInput}.${signature}`;
77
}
78
79
// Exchange parent token for tenant-scoped token
80
async function getTenantToken(tenantOrganizationId) {
81
    const cached = tenantTokenCache.get(tenantOrganizationId);
82
    if (cached && Date.now() < cached.expiresAt - 60000) {
83
        return cached.token;
84
    }
85
86
    const actorToken = await getParentToken();
87
    const subjectToken = createSubjectToken(tenantOrganizationId);
88
89
    const params = new URLSearchParams({
90
        grant_type: 'urn:ietf:params:oauth:grant-type:token-exchange',
91
        subject_token: subjectToken,
92
        subject_token_type: 'urn:ietf:params:oauth:token-type:jwt',
93
        actor_token: actorToken,
94
        actor_token_type: 'urn:ietf:params:oauth:token-type:access_token',
95
    });
96
97
    const response = await axios.post(`${baseURL}/v2/auth/token`, params.toString(), {
98
        headers: {
99
            'Content-Type': 'application/x-www-form-urlencoded',
100
            Authorization: `Bearer ${actorToken}`
101
        }
102
    });
103
104
    tenantTokenCache.set(tenantOrganizationId, {
105
        token: response.data.access_token,
106
        expiresAt: Date.now() + response.data.expires_in * 1000
107
    });
108
109
    return response.data.access_token;
110
}
111
112
// Get all tenants using pagination
113
async function getAllTenants() {
114
    const token = await getParentToken();
115
    let nextPageToken = null;
116
    const allTenants = [];
117
118
    do {
119
        const url = nextPageToken
120
            ? `${baseURL}/v2/tenants?pageToken=${nextPageToken}`
121
            : `${baseURL}/v2/tenants`;
122
123
        const response = await axios.get(url, {
124
            headers: { Authorization: `Bearer ${token}` }
125
        });
126
127
        allTenants.push(...response.data.entries.filter(e => e !== null));
128
        nextPageToken = response.data.nextPageToken || null;
129
    } while (nextPageToken);
130
131
    console.log(`Found ${allTenants.length} tenants`);
132
    return allTenants;
133
}
134
135
// Get connections for a tenant (using tenant-scoped auth token)
136
async function getTenantConnections(tenantOrganizationId) {
137
    const token = await getTenantToken(tenantOrganizationId);
138
    const allConnections = [];
139
    let nextPage = null;
140
141
    do {
142
        const params = new URLSearchParams({ limit: '100' });
143
        if (nextPage) params.set('page', nextPage);
144
145
        const response = await axios.get(`${baseURL}/v2/connections?${params}`, {
146
            headers: { Authorization: `Bearer ${token}` }
147
        });
148
149
        allConnections.push(...response.data.entries);
150
        nextPage = response.data.nextPage || null;
151
    } while (nextPage);
152
153
    return allConnections;
154
}
155
156
// Sync a connection by path (using tenant-scoped token)
157
async function syncConnection(tenantOrganizationId, connectionId, path) {
158
    const token = await getTenantToken(tenantOrganizationId);
159
    const startTime = Date.now();
160
161
    await axios.post(
162
        `${baseURL}/v2/connections/${connectionId}/sync`,
163
        { path },
164
        { headers: { Authorization: `Bearer ${token}`, 'Content-Type': 'application/json' } }
165
    );
166
167
    return { latencyMs: Date.now() - startTime };
168
}
169
170
// Function to process items with controlled concurrency
171
async function processWithConcurrency(items, concurrency, processor) {
172
    const results = [];
173
    const executing = new Set();
174
175
    for (const item of items) {
176
        const promise = processor(item).then(result => {
177
            executing.delete(promise);
178
            return result;
179
        });
180
        executing.add(promise);
181
        results.push(promise);
182
183
        if (executing.size >= concurrency) {
184
            await Promise.race(executing);
185
        }
186
    }
187
188
    return Promise.all(results);
189
}
190
191
// Main function to sync connection paths across all tenants
192
async function syncTenantConnectionPaths() {
193
    const startTime = Date.now();
194
195
    // Step 1: Get all tenants
196
    console.log('\n=== Fetching Tenants ===');
197
    const tenants = await getAllTenants();
198
199
    // Step 2: Process each tenant
200
    console.log('\n=== Syncing Connection Paths ===');
201
    const results = [];
202
203
    await processWithConcurrency(tenants, CONCURRENCY, async (tenant) => {
204
        try {
205
            // Get connections for this tenant
206
            const connections = await getTenantConnections(tenant.organizationId);
207
208
            // Filter by connection name
209
            // Note: similar APIs could be used to retrieve connections via the source swap policy
210
            const targetConnections = CONNECTION_NAME_FILTER
211
                ? connections.filter(c => c.name === CONNECTION_NAME_FILTER)
212
                : connections;
213
214
            if (targetConnections.length === 0) {
215
                results.push({
216
                    tenantName: tenant.name,
217
                    status: 'skipped',
218
                    reason: CONNECTION_NAME_FILTER
219
                        ? `No connection named "${CONNECTION_NAME_FILTER}"`
220
                        : 'No connections found'
221
                });
222
                return;
223
            }
224
225
            // Sync each matching connection
226
            for (const connection of targetConnections) {
227
                try {
228
                    const syncResult = await syncConnection(
229
                        tenant.organizationId,
230
                        connection.connectionId,
231
                        SYNC_PATH
232
                    );
233
                    results.push({
234
                        tenantName: tenant.name,
235
                        connectionName: connection.name,
236
                        status: 'success',
237
                        latencyMs: syncResult.latencyMs
238
                    });
239
                    console.log(`✓ ${tenant.name} - ${connection.name} synced in ${syncResult.latencyMs}ms`);
240
                } catch (error) {
241
                    results.push({
242
                        tenantName: tenant.name,
243
                        connectionName: connection.name,
244
                        status: 'failed',
245
                        error: error.response?.data?.message || error.message
246
                    });
247
                    console.error(`✗ ${tenant.name} - ${connection.name}: ${error.message}`);
248
                }
249
            }
250
        } catch (error) {
251
            results.push({
252
                tenantName: tenant.name,
253
                status: 'failed',
254
                error: error.response?.data?.message || error.message
255
            });
256
            console.error(`✗ ${tenant.name}: ${error.message}`);
257
        }
258
    });
259
260
    // Step 3: Print summary
261
    const totalElapsedMs = Date.now() - startTime;
262
    const successes = results.filter(r => r.status === 'success');
263
    const failures = results.filter(r => r.status === 'failed');
264
    const skipped = results.filter(r => r.status === 'skipped');
265
266
    console.log('\n=== Summary ===');
267
    console.log(`Path synced: ${SYNC_PATH.join('.')}`);
268
    console.log(`Total tenants: ${tenants.length}`);
269
    console.log(`Successful syncs: ${successes.length}`);
270
    console.log(`Failed syncs: ${failures.length}`);
271
    console.log(`Skipped: ${skipped.length}`);
272
    console.log(`Total elapsed time: ${(totalElapsedMs / 1000).toFixed(2)}s`);
273
274
    if (successes.length > 0) {
275
        const latencies = successes.map(r => r.latencyMs).sort((a, b) => a - b);
276
        const avgLatency = latencies.reduce((a, b) => a + b, 0) / latencies.length;
277
        console.log(`\nLatency stats:`);
278
        console.log(`  Min: ${latencies[0]}ms`);
279
        console.log(`  Max: ${latencies[latencies.length - 1]}ms`);
280
        console.log(`  Avg: ${avgLatency.toFixed(0)}ms`);
281
    }
282
283
    if (failures.length > 0) {
284
        console.log('\nFailed syncs:');
285
        console.table(failures.slice(0, 10).map(f => ({
286
            Tenant: f.tenantName,
287
            Connection: f.connectionName || 'N/A',
288
            Error: f.error
289
        })));
290
    }
291
}
292
293
// Execute the main function
294
syncTenantConnectionPaths().catch(error => {
295
    console.error('Failed to sync connection paths:', error);
296
});

1
✓ Tenant A - My Connection synced in 245ms
2
✓ Tenant B - My Connection synced in 198ms
3
✓ Tenant C - My Connection synced in 312ms
4
5
=== Summary ===
6
Path synced: MY_DATABASE.MY_SCHEMA.MY_TABLE
7
Total tenants: 3
8
Successful syncs: 3
9
Failed syncs: 0
10
Skipped: 0
11
Total elapsed time: 1.23s
12
13
Latency stats:
14
  Min: 198ms
15
  Max: 312ms
16
  Avg: 252ms