File size: 3,100 Bytes
a271c58 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 | import { TemporalModule } from 'nestjs-temporal-core';
import { socialIntegrationList } from '@gitroom/nestjs-libraries/integrations/integration.manager';
export const getTemporalModule = (
isWorkers: boolean,
path?: string,
activityClasses?: any[]
) => {
// Queues this worker server should NOT run, comma-separated
// (e.g. EXCLUDE_QUEUE="reddit,x,twitch"). Use it to pin a queue to a single
// server: exclude it on every server except the one that should own it.
// Meant for the providers whose concurrency is too low to split (limit 1).
const excludeQueues = (process.env.EXCLUDE_QUEUE || '')
.split(',')
.map((s) => s.trim())
.filter(Boolean);
// How many worker servers share each (non-excluded) queue. Per-server
// concurrency is divided by this so the GLOBAL concurrency stays correct.
// 1 server => 1 (full), 2 servers => 2 (half each), 3 servers => 3, etc.
const divider = Math.max(
1,
Number(process.env.WORKER_CONCURRENCY_DIVIDER) || 1
);
return TemporalModule.register({
isGlobal: true,
connection: {
address: process.env.TEMPORAL_ADDRESS || 'localhost:7233',
...(process.env.TEMPORAL_TLS === 'true' ? { tls: true } : {}),
...(process.env.TEMPORAL_API_KEY
? { apiKey: process.env.TEMPORAL_API_KEY }
: {}),
namespace: process.env.TEMPORAL_NAMESPACE || 'default',
},
taskQueue: 'main',
logLevel: 'error',
...(isWorkers
? {
workers: [
{ identifier: 'main', maxConcurrentJob: undefined },
...socialIntegrationList,
]
.filter((f) => f.identifier.indexOf('-') === -1)
.map((integration) => ({
integration,
taskQueue: integration.identifier.split('-')[0],
}))
.filter(({ taskQueue }) => !excludeQueues.includes(taskQueue))
.map(({ integration, taskQueue }) => {
// Split the per-provider cap across the servers sharing this
// queue. Floor (never below 1) so the global total never exceeds
// the provider's limit. Providers whose limit is smaller than the
// server count must be pinned via EXCLUDE_QUEUE instead.
const concurrency = integration.maxConcurrentJob
? Math.max(
1,
Math.floor(integration.maxConcurrentJob / divider)
)
: undefined;
return {
taskQueue,
workflowsPath: path!,
activityClasses: activityClasses!,
autoStart: true,
...(concurrency
? {
workerOptions: {
maxConcurrentActivityTaskExecutions: concurrency,
},
}
: {
workerOptions: {
maxConcurrentActivityTaskExecutions: 1000000,
},
}),
};
}),
}
: {}),
});
};
|