mirror of
https://github.com/danny-avila/LibreChat.git
synced 2026-07-10 08:13:46 +00:00
refactor: Extract skill sync trigger orchestrator
This commit is contained in:
Danny Avila
parent
52b100c86c
commit
268a8a21e3
5 changed files with 519 additions and 186 deletions
|
|
@ -1,7 +1,8 @@
|
|||
const { FileContext, skillSyncConfigSchema } = require('librechat-data-provider');
|
||||
const { FileContext } = require('librechat-data-provider');
|
||||
const {
|
||||
getStorageMetadata,
|
||||
createGitHubSkillSyncRunner,
|
||||
createSkillSyncTriggerOrchestrator,
|
||||
startGitHubSkillSyncScheduler,
|
||||
} = require('@librechat/api');
|
||||
const { logger, runAsSystem } = require('@librechat/data-schemas');
|
||||
|
|
@ -11,13 +12,10 @@ const { getStrategyFunctions } = require('~/server/services/Files/strategies');
|
|||
const { getFileStrategy } = require('~/server/utils/getFileStrategy');
|
||||
|
||||
const SYSTEM_USER_ID = '000000000000000000000000';
|
||||
const REQUEST_SYNC_MIN_INTERVAL_MS = 5 * 60 * 1000;
|
||||
const REQUEST_SYNC_STALE_RUNNING_MS = 35 * 60 * 1000;
|
||||
|
||||
let appConfigRef;
|
||||
let runner;
|
||||
let scheduler;
|
||||
const requestSyncsInFlight = new Set();
|
||||
|
||||
async function loadCurrentAppConfig() {
|
||||
try {
|
||||
|
|
@ -150,155 +148,17 @@ function createRunner({ getConfig, loadAppConfig } = {}) {
|
|||
};
|
||||
}
|
||||
|
||||
function parseSkillSyncConfig(raw) {
|
||||
if (!raw || typeof raw !== 'object') {
|
||||
return undefined;
|
||||
}
|
||||
const parsed = skillSyncConfigSchema.safeParse(raw);
|
||||
if (!parsed.success) {
|
||||
logger.warn('[GitHubSkillSync] Ignoring invalid request-scoped skill sync config', {
|
||||
issues: parsed.error.flatten(),
|
||||
});
|
||||
return undefined;
|
||||
}
|
||||
return parsed.data;
|
||||
}
|
||||
|
||||
function isSameSkillSyncConfig(left, right) {
|
||||
return JSON.stringify(left ?? null) === JSON.stringify(right ?? null);
|
||||
}
|
||||
|
||||
function getRequestTenantId(user) {
|
||||
return typeof user?.tenantId === 'string' && user.tenantId ? user.tenantId : undefined;
|
||||
}
|
||||
|
||||
function withRequestTenant(config, user, { disableRunOnStartup = false } = {}) {
|
||||
const tenantId = getRequestTenantId(user);
|
||||
return {
|
||||
...config,
|
||||
github: {
|
||||
...config.github,
|
||||
...(disableRunOnStartup ? { runOnStartup: false } : {}),
|
||||
sources: config.github.sources.map((source) => ({
|
||||
...source,
|
||||
tenantId,
|
||||
})),
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function getRequestSkillSyncConfig(appConfig, user) {
|
||||
const resolved = parseSkillSyncConfig(appConfig?.skillSync);
|
||||
if (!resolved?.github?.enabled || resolved.github.sources.length === 0) {
|
||||
return undefined;
|
||||
}
|
||||
|
||||
const base = parseSkillSyncConfig(appConfig?.config?.skillSync);
|
||||
if (isSameSkillSyncConfig(resolved, base)) {
|
||||
return undefined;
|
||||
}
|
||||
|
||||
return withRequestTenant(resolved, user, { disableRunOnStartup: true });
|
||||
}
|
||||
|
||||
function getAdminRequestSkillSyncConfig(appConfig, user) {
|
||||
const resolved = parseSkillSyncConfig(appConfig?.skillSync);
|
||||
if (!resolved?.github) {
|
||||
return resolved;
|
||||
}
|
||||
|
||||
const base = parseSkillSyncConfig(appConfig?.config?.skillSync);
|
||||
if (isSameSkillSyncConfig(resolved, base)) {
|
||||
return resolved;
|
||||
}
|
||||
|
||||
return withRequestTenant(resolved, user);
|
||||
}
|
||||
|
||||
function toTimestamp(value) {
|
||||
if (!value) {
|
||||
return 0;
|
||||
}
|
||||
if (value instanceof Date) {
|
||||
return value.getTime();
|
||||
}
|
||||
const parsed = Date.parse(String(value));
|
||||
return Number.isNaN(parsed) ? 0 : parsed;
|
||||
}
|
||||
|
||||
function getLastAttemptAt(source) {
|
||||
return Math.max(
|
||||
toTimestamp(source.finishedAt),
|
||||
toTimestamp(source.startedAt),
|
||||
toTimestamp(source.updatedAt),
|
||||
toTimestamp(source.lastSuccessAt),
|
||||
toTimestamp(source.lastFailureAt),
|
||||
);
|
||||
}
|
||||
|
||||
function shouldRunRequestSync(status) {
|
||||
if (!status.enabled || status.sources.length === 0) {
|
||||
return false;
|
||||
}
|
||||
const intervalMs = Math.max(
|
||||
REQUEST_SYNC_MIN_INTERVAL_MS,
|
||||
(status.intervalMinutes ?? 60) * 60 * 1000,
|
||||
);
|
||||
const now = Date.now();
|
||||
return status.sources.some((source) => {
|
||||
if (source.status === 'running') {
|
||||
const startedAt = toTimestamp(source.startedAt);
|
||||
return Boolean(startedAt && now - startedAt >= REQUEST_SYNC_STALE_RUNNING_MS);
|
||||
}
|
||||
const lastAttemptAt = getLastAttemptAt(source);
|
||||
return !lastAttemptAt || now - lastAttemptAt >= intervalMs;
|
||||
});
|
||||
}
|
||||
const triggerOrchestrator = createSkillSyncTriggerOrchestrator({
|
||||
createRunner,
|
||||
logger,
|
||||
});
|
||||
|
||||
function getGitHubSkillSyncRunnerForRequest(req) {
|
||||
const config = getAdminRequestSkillSyncConfig(req.config, req.user);
|
||||
return createRunner({
|
||||
getConfig: async () => config,
|
||||
loadAppConfig: async () => req.config,
|
||||
});
|
||||
}
|
||||
|
||||
function getRequestSyncKey(config, user) {
|
||||
const tenantId = user?.tenantId ?? '';
|
||||
const sources = config.github.sources
|
||||
.map((source) => source.id)
|
||||
.sort()
|
||||
.join(',');
|
||||
return `${tenantId}:${sources}`;
|
||||
return triggerOrchestrator.getRunnerForAdminRequest(req);
|
||||
}
|
||||
|
||||
async function maybeRunGitHubSkillSyncForRequest(req) {
|
||||
const config = getRequestSkillSyncConfig(req.config, req.user);
|
||||
if (!config) {
|
||||
return false;
|
||||
}
|
||||
|
||||
const syncKey = getRequestSyncKey(config, req.user);
|
||||
if (requestSyncsInFlight.has(syncKey)) {
|
||||
return false;
|
||||
}
|
||||
|
||||
const loadAppConfig = async () => req.config;
|
||||
const requestRunner = createRunner({
|
||||
getConfig: async () => config,
|
||||
loadAppConfig,
|
||||
});
|
||||
const status = await requestRunner.getStatus();
|
||||
if (!shouldRunRequestSync(status)) {
|
||||
return false;
|
||||
}
|
||||
|
||||
requestSyncsInFlight.add(syncKey);
|
||||
void requestRunner
|
||||
.runOnce()
|
||||
.catch((error) => logger.error('[GitHubSkillSync] Request-scoped sync failed:', error))
|
||||
.finally(() => requestSyncsInFlight.delete(syncKey));
|
||||
return true;
|
||||
return triggerOrchestrator.maybeRunForRequest(req);
|
||||
}
|
||||
|
||||
function initializeGitHubSkillSync(appConfig) {
|
||||
|
|
|
|||
|
|
@ -11,44 +11,48 @@ jest.mock('~/server/services/Config', () => ({
|
|||
getAppConfig: mockGetAppConfig,
|
||||
}));
|
||||
|
||||
jest.mock('@librechat/api', () => ({
|
||||
createGitHubSkillSyncRunner: jest.fn((deps) => {
|
||||
mockRunnerDeps = deps;
|
||||
const runner = {
|
||||
getStatus: jest.fn(async () => {
|
||||
if (mockRunnerStatus) {
|
||||
return mockRunnerStatus;
|
||||
}
|
||||
const config = await deps.getConfig();
|
||||
const github = config?.github ?? {};
|
||||
return {
|
||||
enabled: github.enabled ?? false,
|
||||
intervalMinutes: github.intervalMinutes ?? 60,
|
||||
runOnStartup: github.runOnStartup ?? false,
|
||||
sources: (github.sources ?? []).map((source) => ({
|
||||
provider: 'github',
|
||||
sourceId: source.id,
|
||||
status: 'idle',
|
||||
owner: source.owner,
|
||||
repo: source.repo,
|
||||
ref: source.ref,
|
||||
paths: source.paths,
|
||||
syncedSkillCount: 0,
|
||||
syncedFileCount: 0,
|
||||
deletedSkillCount: 0,
|
||||
deletedFileCount: 0,
|
||||
})),
|
||||
credentials: [],
|
||||
};
|
||||
}),
|
||||
runOnce: jest.fn(async () => deps.getConfig()),
|
||||
};
|
||||
mockCreatedRunners.push({ deps, runner });
|
||||
return runner;
|
||||
}),
|
||||
getStorageMetadata: jest.fn(() => ({})),
|
||||
startGitHubSkillSyncScheduler: jest.fn(() => ({ stop: jest.fn() })),
|
||||
}));
|
||||
jest.mock('@librechat/api', () => {
|
||||
const actualApi = jest.requireActual('@librechat/api');
|
||||
return {
|
||||
createSkillSyncTriggerOrchestrator: actualApi.createSkillSyncTriggerOrchestrator,
|
||||
createGitHubSkillSyncRunner: jest.fn((deps) => {
|
||||
mockRunnerDeps = deps;
|
||||
const runner = {
|
||||
getStatus: jest.fn(async () => {
|
||||
if (mockRunnerStatus) {
|
||||
return mockRunnerStatus;
|
||||
}
|
||||
const config = await deps.getConfig();
|
||||
const github = config?.github ?? {};
|
||||
return {
|
||||
enabled: github.enabled ?? false,
|
||||
intervalMinutes: github.intervalMinutes ?? 60,
|
||||
runOnStartup: github.runOnStartup ?? false,
|
||||
sources: (github.sources ?? []).map((source) => ({
|
||||
provider: 'github',
|
||||
sourceId: source.id,
|
||||
status: 'idle',
|
||||
owner: source.owner,
|
||||
repo: source.repo,
|
||||
ref: source.ref,
|
||||
paths: source.paths,
|
||||
syncedSkillCount: 0,
|
||||
syncedFileCount: 0,
|
||||
deletedSkillCount: 0,
|
||||
deletedFileCount: 0,
|
||||
})),
|
||||
credentials: [],
|
||||
};
|
||||
}),
|
||||
runOnce: jest.fn(async () => deps.getConfig()),
|
||||
};
|
||||
mockCreatedRunners.push({ deps, runner });
|
||||
return runner;
|
||||
}),
|
||||
getStorageMetadata: jest.fn(() => ({})),
|
||||
startGitHubSkillSyncScheduler: jest.fn(() => ({ stop: jest.fn() })),
|
||||
};
|
||||
});
|
||||
|
||||
jest.mock('@librechat/data-schemas', () => ({
|
||||
logger: {
|
||||
|
|
|
|||
|
|
@ -1,2 +1,3 @@
|
|||
export * from './github';
|
||||
export * from './orchestrator';
|
||||
export * from './scheduler';
|
||||
|
|
|
|||
228
packages/api/src/skills/sync/orchestrator.spec.ts
Normal file
228
packages/api/src/skills/sync/orchestrator.spec.ts
Normal file
|
|
@ -0,0 +1,228 @@
|
|||
import type { SkillSyncConfig } from 'librechat-data-provider';
|
||||
import type { GitHubSkillSyncRunner } from './github';
|
||||
import type { SkillSyncTriggerRunnerFactoryInput } from './orchestrator';
|
||||
import { createSkillSyncTriggerOrchestrator } from './orchestrator';
|
||||
|
||||
type RunnerStatus = Awaited<ReturnType<GitHubSkillSyncRunner['getStatus']>>;
|
||||
type RunnerRunResult = Awaited<ReturnType<GitHubSkillSyncRunner['runOnce']>>;
|
||||
|
||||
const source = {
|
||||
id: 'tenant-skills',
|
||||
owner: 'LibreChat',
|
||||
repo: 'skills',
|
||||
ref: 'main',
|
||||
paths: ['skills'],
|
||||
token: '${GITHUB_SKILLS_TOKEN}',
|
||||
tenantId: 'other-tenant',
|
||||
};
|
||||
|
||||
function skillSync(
|
||||
overrides: Partial<NonNullable<SkillSyncConfig>['github']> = {},
|
||||
): SkillSyncConfig {
|
||||
return {
|
||||
github: {
|
||||
enabled: true,
|
||||
intervalMinutes: 60,
|
||||
runOnStartup: true,
|
||||
sources: [source],
|
||||
...overrides,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function statusFromConfig(config: SkillSyncConfig | undefined): RunnerStatus {
|
||||
const github = config?.github;
|
||||
return {
|
||||
enabled: github?.enabled ?? false,
|
||||
intervalMinutes: github?.intervalMinutes ?? 60,
|
||||
runOnStartup: github?.runOnStartup ?? false,
|
||||
sources:
|
||||
github?.sources.map((configuredSource) => ({
|
||||
provider: 'github',
|
||||
sourceId: configuredSource.id,
|
||||
tenantId: configuredSource.tenantId,
|
||||
status: 'idle',
|
||||
credentialKey: configuredSource.credentialKey,
|
||||
credentialPresent: Boolean(configuredSource.credentialKey || configuredSource.token),
|
||||
owner: configuredSource.owner,
|
||||
repo: configuredSource.repo,
|
||||
ref: configuredSource.ref,
|
||||
paths: configuredSource.paths,
|
||||
syncedSkillCount: 0,
|
||||
syncedFileCount: 0,
|
||||
deletedSkillCount: 0,
|
||||
deletedFileCount: 0,
|
||||
})) ?? [],
|
||||
credentials: [],
|
||||
fineGrainedTokenRecommendation: 'Use a GitHub fine-grained personal access token.',
|
||||
};
|
||||
}
|
||||
|
||||
function completedRun(): RunnerRunResult {
|
||||
return { status: 'completed', sources: [] };
|
||||
}
|
||||
|
||||
function deferred<T>() {
|
||||
let resolve!: (value: T) => void;
|
||||
const promise = new Promise<T>((res) => {
|
||||
resolve = res;
|
||||
});
|
||||
return { promise, resolve };
|
||||
}
|
||||
|
||||
async function flushPromises() {
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
}
|
||||
|
||||
function createHarness(
|
||||
options: {
|
||||
status?: RunnerStatus;
|
||||
runOnce?: () => Promise<RunnerRunResult>;
|
||||
} = {},
|
||||
) {
|
||||
const runners: Array<{
|
||||
input: SkillSyncTriggerRunnerFactoryInput;
|
||||
runner: GitHubSkillSyncRunner;
|
||||
}> = [];
|
||||
const logger = {
|
||||
warn: jest.fn(),
|
||||
error: jest.fn(),
|
||||
};
|
||||
const createRunner = jest.fn((input: SkillSyncTriggerRunnerFactoryInput) => {
|
||||
const runner: GitHubSkillSyncRunner = {
|
||||
getStatus: jest.fn(async () => options.status ?? statusFromConfig(await input.getConfig())),
|
||||
runOnce: jest.fn(options.runOnce ?? (async () => completedRun())),
|
||||
};
|
||||
runners.push({ input, runner });
|
||||
return runner;
|
||||
});
|
||||
const orchestrator = createSkillSyncTriggerOrchestrator({
|
||||
createRunner,
|
||||
logger,
|
||||
});
|
||||
return { createRunner, logger, orchestrator, runners };
|
||||
}
|
||||
|
||||
describe('createSkillSyncTriggerOrchestrator', () => {
|
||||
it('starts request sync from resolved admin skillSync config and derives tenant from the request', async () => {
|
||||
const config = skillSync();
|
||||
const { orchestrator, runners } = createHarness();
|
||||
|
||||
const started = await orchestrator.maybeRunForRequest({
|
||||
config: { skillSync: config, config: {} },
|
||||
user: { tenantId: 'tenant-a' },
|
||||
});
|
||||
|
||||
const requestConfig = await runners[0].input.getConfig();
|
||||
expect(started).toBe(true);
|
||||
expect(runners[0].runner.runOnce).toHaveBeenCalledTimes(1);
|
||||
expect(requestConfig?.github?.runOnStartup).toBe(false);
|
||||
expect(requestConfig?.github?.sources[0]).toEqual(
|
||||
expect.objectContaining({ id: 'tenant-skills', tenantId: 'tenant-a' }),
|
||||
);
|
||||
});
|
||||
|
||||
it('does not start request sync for base YAML skillSync config', async () => {
|
||||
const config = skillSync();
|
||||
const { createRunner, orchestrator } = createHarness();
|
||||
|
||||
const started = await orchestrator.maybeRunForRequest({
|
||||
config: { skillSync: config, config: { skillSync: config } },
|
||||
user: { tenantId: 'tenant-a' },
|
||||
});
|
||||
|
||||
expect(started).toBe(false);
|
||||
expect(createRunner).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('creates an admin request runner from resolved config without disabling startup runs', async () => {
|
||||
const config = skillSync();
|
||||
const { orchestrator, runners } = createHarness();
|
||||
|
||||
const runner = orchestrator.getRunnerForAdminRequest({
|
||||
config: { skillSync: config, config: {} },
|
||||
user: { tenantId: 'tenant-a' },
|
||||
});
|
||||
const runnerConfig = await runners[0].input.getConfig();
|
||||
|
||||
expect(runner).toBe(runners[0].runner);
|
||||
expect(runnerConfig?.github?.runOnStartup).toBe(true);
|
||||
expect(runnerConfig?.github?.sources[0]).toEqual(
|
||||
expect.objectContaining({ id: 'tenant-skills', tenantId: 'tenant-a' }),
|
||||
);
|
||||
});
|
||||
|
||||
it('does not start request sync when the configured source is already running', async () => {
|
||||
const config = skillSync({ runOnStartup: false });
|
||||
const { orchestrator, runners } = createHarness({
|
||||
status: {
|
||||
...statusFromConfig(config),
|
||||
sources: [
|
||||
{
|
||||
...statusFromConfig(config).sources[0],
|
||||
status: 'running',
|
||||
startedAt: new Date(),
|
||||
},
|
||||
],
|
||||
},
|
||||
});
|
||||
|
||||
const started = await orchestrator.maybeRunForRequest({
|
||||
config: { skillSync: config, config: {} },
|
||||
user: { tenantId: 'tenant-a' },
|
||||
});
|
||||
|
||||
expect(started).toBe(false);
|
||||
expect(runners[0].runner.runOnce).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('retries request sync when a running source status is stale', async () => {
|
||||
const config = skillSync({ runOnStartup: false });
|
||||
const { orchestrator, runners } = createHarness({
|
||||
status: {
|
||||
...statusFromConfig(config),
|
||||
sources: [
|
||||
{
|
||||
...statusFromConfig(config).sources[0],
|
||||
status: 'running',
|
||||
startedAt: new Date(Date.now() - 40 * 60 * 1000),
|
||||
},
|
||||
],
|
||||
},
|
||||
});
|
||||
|
||||
const started = await orchestrator.maybeRunForRequest({
|
||||
config: { skillSync: config, config: {} },
|
||||
user: { tenantId: 'tenant-a' },
|
||||
});
|
||||
|
||||
expect(started).toBe(true);
|
||||
expect(runners[0].runner.runOnce).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('suppresses duplicate request sync while an equivalent run is in flight', async () => {
|
||||
const pendingRun = deferred<RunnerRunResult>();
|
||||
const config = skillSync({ runOnStartup: false });
|
||||
const { orchestrator, runners } = createHarness({
|
||||
runOnce: () => pendingRun.promise,
|
||||
});
|
||||
|
||||
const first = await orchestrator.maybeRunForRequest({
|
||||
config: { skillSync: config, config: {} },
|
||||
user: { tenantId: 'tenant-a' },
|
||||
});
|
||||
const second = await orchestrator.maybeRunForRequest({
|
||||
config: { skillSync: config, config: {} },
|
||||
user: { tenantId: 'tenant-a' },
|
||||
});
|
||||
|
||||
pendingRun.resolve(completedRun());
|
||||
await flushPromises();
|
||||
|
||||
expect(first).toBe(true);
|
||||
expect(second).toBe(false);
|
||||
expect(runners).toHaveLength(1);
|
||||
expect(runners[0].runner.runOnce).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
});
|
||||
240
packages/api/src/skills/sync/orchestrator.ts
Normal file
240
packages/api/src/skills/sync/orchestrator.ts
Normal file
|
|
@ -0,0 +1,240 @@
|
|||
import { skillSyncConfigSchema } from 'librechat-data-provider';
|
||||
import type { SkillSyncConfig } from 'librechat-data-provider';
|
||||
import type { GitHubSkillSyncRunner } from './github';
|
||||
|
||||
const REQUEST_SYNC_MIN_INTERVAL_MS = 5 * 60 * 1000;
|
||||
const REQUEST_SYNC_STALE_RUNNING_MS = 35 * 60 * 1000;
|
||||
|
||||
type MaybePromise<T> = T | Promise<T>;
|
||||
|
||||
type SkillSyncAppConfigLike = {
|
||||
skillSync?: unknown;
|
||||
config?: {
|
||||
skillSync?: unknown;
|
||||
};
|
||||
};
|
||||
|
||||
type SkillSyncRequestUser = {
|
||||
tenantId?: string | null;
|
||||
};
|
||||
|
||||
type SkillSyncRequestLike = {
|
||||
config?: SkillSyncAppConfigLike;
|
||||
user?: SkillSyncRequestUser;
|
||||
};
|
||||
|
||||
type ResolvedSkillSyncConfig = NonNullable<SkillSyncConfig>;
|
||||
type ResolvedGitHubSkillSyncConfig = NonNullable<ResolvedSkillSyncConfig['github']>;
|
||||
type SkillSyncConfigWithGitHub = ResolvedSkillSyncConfig & {
|
||||
github: ResolvedGitHubSkillSyncConfig;
|
||||
};
|
||||
|
||||
type SkillSyncRunnerStatus = Awaited<ReturnType<GitHubSkillSyncRunner['getStatus']>>;
|
||||
|
||||
type SkillSyncTriggerLogger = {
|
||||
warn: (message: string, metadata?: object) => void;
|
||||
error: (message: string, error?: unknown) => void;
|
||||
};
|
||||
|
||||
export type SkillSyncTriggerRunnerFactoryInput = {
|
||||
getConfig: () => MaybePromise<SkillSyncConfig | undefined>;
|
||||
loadAppConfig: () => MaybePromise<SkillSyncAppConfigLike | undefined>;
|
||||
};
|
||||
|
||||
export type SkillSyncTriggerOrchestratorDeps = {
|
||||
createRunner: (input: SkillSyncTriggerRunnerFactoryInput) => GitHubSkillSyncRunner;
|
||||
logger: SkillSyncTriggerLogger;
|
||||
minIntervalMs?: number;
|
||||
staleRunningMs?: number;
|
||||
inFlight?: Set<string>;
|
||||
};
|
||||
|
||||
function parseSkillSyncConfig(
|
||||
raw: unknown,
|
||||
logger: SkillSyncTriggerLogger,
|
||||
): SkillSyncConfig | undefined {
|
||||
if (!raw || typeof raw !== 'object') {
|
||||
return undefined;
|
||||
}
|
||||
const parsed = skillSyncConfigSchema.safeParse(raw);
|
||||
if (!parsed.success) {
|
||||
logger.warn('[GitHubSkillSync] Ignoring invalid skill sync config', {
|
||||
issues: parsed.error.flatten(),
|
||||
});
|
||||
return undefined;
|
||||
}
|
||||
return parsed.data;
|
||||
}
|
||||
|
||||
function hasGitHubConfig(config: SkillSyncConfig | undefined): config is SkillSyncConfigWithGitHub {
|
||||
return Boolean(config?.github);
|
||||
}
|
||||
|
||||
function isSameSkillSyncConfig(
|
||||
left: SkillSyncConfig | undefined,
|
||||
right: SkillSyncConfig | undefined,
|
||||
) {
|
||||
return JSON.stringify(left ?? null) === JSON.stringify(right ?? null);
|
||||
}
|
||||
|
||||
function getRequestTenantId(user: SkillSyncRequestUser | undefined): string | undefined {
|
||||
return typeof user?.tenantId === 'string' && user.tenantId ? user.tenantId : undefined;
|
||||
}
|
||||
|
||||
function withRequestTenant(
|
||||
config: SkillSyncConfigWithGitHub,
|
||||
user: SkillSyncRequestUser | undefined,
|
||||
{ disableRunOnStartup = false } = {},
|
||||
): SkillSyncConfig {
|
||||
const tenantId = getRequestTenantId(user);
|
||||
return {
|
||||
...config,
|
||||
github: {
|
||||
...config.github,
|
||||
...(disableRunOnStartup ? { runOnStartup: false } : {}),
|
||||
sources: config.github.sources.map((source) => ({
|
||||
...source,
|
||||
tenantId,
|
||||
})),
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function getRequestSkillSyncConfig(
|
||||
appConfig: SkillSyncAppConfigLike | undefined,
|
||||
user: SkillSyncRequestUser | undefined,
|
||||
logger: SkillSyncTriggerLogger,
|
||||
): SkillSyncConfig | undefined {
|
||||
const resolved = parseSkillSyncConfig(appConfig?.skillSync, logger);
|
||||
if (
|
||||
!hasGitHubConfig(resolved) ||
|
||||
!resolved.github.enabled ||
|
||||
resolved.github.sources.length === 0
|
||||
) {
|
||||
return undefined;
|
||||
}
|
||||
|
||||
const base = parseSkillSyncConfig(appConfig?.config?.skillSync, logger);
|
||||
if (isSameSkillSyncConfig(resolved, base)) {
|
||||
return undefined;
|
||||
}
|
||||
|
||||
return withRequestTenant(resolved, user, { disableRunOnStartup: true });
|
||||
}
|
||||
|
||||
function getAdminRequestSkillSyncConfig(
|
||||
appConfig: SkillSyncAppConfigLike | undefined,
|
||||
user: SkillSyncRequestUser | undefined,
|
||||
logger: SkillSyncTriggerLogger,
|
||||
): SkillSyncConfig | undefined {
|
||||
const resolved = parseSkillSyncConfig(appConfig?.skillSync, logger);
|
||||
if (!hasGitHubConfig(resolved)) {
|
||||
return resolved;
|
||||
}
|
||||
|
||||
const base = parseSkillSyncConfig(appConfig?.config?.skillSync, logger);
|
||||
if (isSameSkillSyncConfig(resolved, base)) {
|
||||
return resolved;
|
||||
}
|
||||
|
||||
return withRequestTenant(resolved, user);
|
||||
}
|
||||
|
||||
function toTimestamp(value: unknown): number {
|
||||
if (!value) {
|
||||
return 0;
|
||||
}
|
||||
if (value instanceof Date) {
|
||||
return value.getTime();
|
||||
}
|
||||
const parsed = Date.parse(String(value));
|
||||
return Number.isNaN(parsed) ? 0 : parsed;
|
||||
}
|
||||
|
||||
function getLastAttemptAt(source: SkillSyncRunnerStatus['sources'][number]): number {
|
||||
return Math.max(
|
||||
toTimestamp(source.finishedAt),
|
||||
toTimestamp(source.startedAt),
|
||||
toTimestamp(source.updatedAt),
|
||||
toTimestamp(source.lastSuccessAt),
|
||||
toTimestamp(source.lastFailureAt),
|
||||
);
|
||||
}
|
||||
|
||||
function shouldRunRequestSync(
|
||||
status: SkillSyncRunnerStatus,
|
||||
{ minIntervalMs, staleRunningMs }: { minIntervalMs: number; staleRunningMs: number },
|
||||
): boolean {
|
||||
if (!status.enabled || status.sources.length === 0) {
|
||||
return false;
|
||||
}
|
||||
const intervalMs = Math.max(minIntervalMs, (status.intervalMinutes ?? 60) * 60 * 1000);
|
||||
const now = Date.now();
|
||||
return status.sources.some((source) => {
|
||||
if (source.status === 'running') {
|
||||
const startedAt = toTimestamp(source.startedAt);
|
||||
return Boolean(startedAt && now - startedAt >= staleRunningMs);
|
||||
}
|
||||
const lastAttemptAt = getLastAttemptAt(source);
|
||||
return !lastAttemptAt || now - lastAttemptAt >= intervalMs;
|
||||
});
|
||||
}
|
||||
|
||||
function getRequestSyncKey(
|
||||
config: SkillSyncConfigWithGitHub,
|
||||
user: SkillSyncRequestUser | undefined,
|
||||
) {
|
||||
const tenantId = user?.tenantId ?? '';
|
||||
const sources = config.github.sources
|
||||
.map((source) => source.id)
|
||||
.sort()
|
||||
.join(',');
|
||||
return `${tenantId}:${sources}`;
|
||||
}
|
||||
|
||||
export function createSkillSyncTriggerOrchestrator(deps: SkillSyncTriggerOrchestratorDeps) {
|
||||
const inFlight = deps.inFlight ?? new Set<string>();
|
||||
const minIntervalMs = deps.minIntervalMs ?? REQUEST_SYNC_MIN_INTERVAL_MS;
|
||||
const staleRunningMs = deps.staleRunningMs ?? REQUEST_SYNC_STALE_RUNNING_MS;
|
||||
|
||||
function getRunnerForAdminRequest(request: SkillSyncRequestLike): GitHubSkillSyncRunner {
|
||||
const config = getAdminRequestSkillSyncConfig(request.config, request.user, deps.logger);
|
||||
return deps.createRunner({
|
||||
getConfig: async () => config,
|
||||
loadAppConfig: async () => request.config,
|
||||
});
|
||||
}
|
||||
|
||||
async function maybeRunForRequest(request: SkillSyncRequestLike): Promise<boolean> {
|
||||
const config = getRequestSkillSyncConfig(request.config, request.user, deps.logger);
|
||||
if (!hasGitHubConfig(config)) {
|
||||
return false;
|
||||
}
|
||||
|
||||
const syncKey = getRequestSyncKey(config, request.user);
|
||||
if (inFlight.has(syncKey)) {
|
||||
return false;
|
||||
}
|
||||
|
||||
const requestRunner = deps.createRunner({
|
||||
getConfig: async () => config,
|
||||
loadAppConfig: async () => request.config,
|
||||
});
|
||||
const status = await requestRunner.getStatus();
|
||||
if (!shouldRunRequestSync(status, { minIntervalMs, staleRunningMs })) {
|
||||
return false;
|
||||
}
|
||||
|
||||
inFlight.add(syncKey);
|
||||
void requestRunner
|
||||
.runOnce()
|
||||
.catch((error) => deps.logger.error('[GitHubSkillSync] Request-scoped sync failed:', error))
|
||||
.finally(() => inFlight.delete(syncKey));
|
||||
return true;
|
||||
}
|
||||
|
||||
return {
|
||||
getRunnerForAdminRequest,
|
||||
maybeRunForRequest,
|
||||
};
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue