feat: rebuild warm operations workbench

This commit is contained in:
Codex
2026-07-30 14:49:33 +08:00
parent 25c853dea8
commit 570edf6bb2
88 changed files with 13413 additions and 2250 deletions
+1
View File
@@ -16,6 +16,7 @@
"@multi-simadmin/contracts": "workspace:*",
"@multi-simadmin/operation-registry": "workspace:*",
"better-sqlite3": "12.11.1",
"cron-parser": "^5.6.2",
"drizzle-orm": "0.45.2",
"fastify": "5.10.0",
"tsx": "4.22.4"
@@ -0,0 +1,47 @@
import { describe, expect, it } from 'vitest';
import { nextOccurrence, previewCron, reconcileOccurrence } from './schedule-time.js';
describe('schedule time', () => {
it('calculates future occurrences in Beijing time across the UTC day boundary', () => {
expect(previewCron('0 9 * * *', 2, new Date('2026-07-30T00:30:00.000Z'))).toEqual([
'2026-07-30T01:00:00.000Z',
'2026-07-31T01:00:00.000Z',
]);
expect(nextOccurrence('30 0 * * *', new Date('2026-07-30T15:59:00.000Z'))).toBe(
'2026-07-30T16:30:00.000Z',
);
});
it('rejects seconds fields and invalid cron grammar', () => {
expect(() => previewCron('* * * * * *', 1)).toThrow(/five-field/i);
expect(() => previewCron('99 99 * * *', 1)).toThrow(/cron/i);
});
it('reconciles a missed due time according to the per-task policy', () => {
const common = {
cronExpression: '*/10 * * * *',
nextDueAt: '2026-07-30T01:00:00.000Z',
};
expect(
reconcileOccurrence(
{ ...common, misfirePolicy: 'skip' },
new Date('2026-07-30T01:25:00.000Z'),
),
).toEqual({
action: 'skip',
dueAt: '2026-07-30T01:20:00.000Z',
nextDueAt: '2026-07-30T01:30:00.000Z',
});
expect(
reconcileOccurrence(
{ ...common, misfirePolicy: 'catch-up-once' },
new Date('2026-07-30T01:25:00.000Z'),
),
).toEqual({
action: 'run',
dueAt: '2026-07-30T01:20:00.000Z',
nextDueAt: '2026-07-30T01:30:00.000Z',
});
});
});
@@ -0,0 +1,65 @@
import type { ScheduleMisfirePolicy } from '@multi-simadmin/contracts';
import { CronExpressionParser } from 'cron-parser';
const TIMEZONE = 'Asia/Shanghai';
function validateExpression(expression: string): string {
const clean = expression.trim().replace(/\s+/g, ' ');
if (clean.split(' ').length !== 5)
throw new TypeError('Cron must use standard five-field syntax');
return clean;
}
function parser(expression: string, currentDate: Date) {
try {
return CronExpressionParser.parse(validateExpression(expression), {
currentDate,
tz: TIMEZONE,
});
} catch (error) {
if (error instanceof TypeError && /five-field/.test(error.message)) throw error;
throw new TypeError('Cron expression is invalid');
}
}
export function previewCron(
expression: string,
count: number,
from = new Date(),
): readonly string[] {
if (!Number.isSafeInteger(count) || count < 1 || count > 10)
throw new TypeError('Cron preview count must be between 1 and 10');
const interval = parser(expression, from);
return Array.from({ length: count }, () => interval.next().toDate().toISOString());
}
export function nextOccurrence(expression: string, after: Date): string {
return parser(expression, after).next().toDate().toISOString();
}
export type ReconciledOccurrence =
| { readonly action: 'wait'; readonly dueAt: string; readonly nextDueAt: string }
| { readonly action: 'run' | 'skip'; readonly dueAt: string; readonly nextDueAt: string };
export function reconcileOccurrence(
task: {
readonly cronExpression: string;
readonly nextDueAt: string;
readonly misfirePolicy: ScheduleMisfirePolicy;
},
now: Date,
): ReconciledOccurrence {
const nextDue = new Date(task.nextDueAt);
if (!Number.isFinite(nextDue.getTime())) throw new TypeError('nextDueAt is invalid');
if (nextDue.getTime() > now.getTime())
return { action: 'wait', dueAt: task.nextDueAt, nextDueAt: task.nextDueAt };
const interval = parser(task.cronExpression, now);
const dueAt = interval.prev().toDate().toISOString();
const nextDueAt = nextOccurrence(task.cronExpression, now);
return {
action: task.misfirePolicy === 'catch-up-once' ? 'run' : 'skip',
dueAt,
nextDueAt,
};
}
@@ -0,0 +1,125 @@
import Database from 'better-sqlite3';
import { describe, expect, it, vi } from 'vitest';
import { migrateDatabase } from '../../infrastructure/database/migrations.js';
import { ScheduledTaskRepository } from './scheduled-task-repository.js';
import { ScheduledOperationDispatcher } from './scheduled-operation-dispatcher.js';
describe('ScheduledOperationDispatcher', () => {
it('executes restart targets independently and aggregates partial success', async () => {
const db = new Database(':memory:');
migrateDatabase(db);
const execute = vi
.fn()
.mockResolvedValueOnce({ id: 'job-a', status: 'succeeded' })
.mockResolvedValueOnce({ id: 'job-b', status: 'failed' });
const dispatcher = new ScheduledOperationDispatcher({
db,
operations: {
prepare: vi.fn(async (input) => ({
id: `prep-${input.targets[0]?.instanceId}`,
confirmationToken: 'token',
})),
execute,
},
messages: { send: vi.fn() },
store: { set: vi.fn(), get: vi.fn(), delete: vi.fn() },
repository: new ScheduledTaskRepository(db),
});
const result = await dispatcher.dispatch(
{
id: 'task-1',
name: 'Restart',
operationType: 'restart-service',
cronExpression: '0 9 * * *',
timezone: 'Asia/Shanghai',
targetSelector: { mode: 'fixed', instanceIds: ['a', 'b'] },
misfirePolicy: 'skip',
overlapPolicy: 'skip',
retryPolicy: { maxRetries: 0, intervalSeconds: 60 },
enabled: true,
version: 1,
createdBy: 'operator',
updatedBy: 'operator',
createdAt: '2026-07-30T00:00:00.000Z',
updatedAt: '2026-07-30T00:00:00.000Z',
},
[
{ id: 'a', revision: 1 },
{ id: 'b', revision: 2 },
],
{ actor: 'operator', requestId: 'request-1' },
);
expect(result).toEqual({ outcome: 'partially-succeeded', jobIds: ['job-a', 'job-b'] });
expect(execute).toHaveBeenCalledTimes(2);
db.close();
});
it('retries only failed SMS recipients using the task retry interval', async () => {
const db = new Database(':memory:');
migrateDatabase(db);
db.prepare(
'INSERT INTO instances (id,name,base_url,enabled,config_revision,created_at,updated_at) VALUES (?,?,?,?,?,?,?)',
).run(
'a',
'Alpha',
'http://10.0.0.1',
1,
1,
'2026-07-30T00:00:00.000Z',
'2026-07-30T00:00:00.000Z',
);
const repository = new ScheduledTaskRepository(db);
const task = repository.create({
id: 'task-sms',
task: {
name: 'Notify lab',
operationType: 'send-sms',
cronExpression: '0 9 * * *',
timezone: 'Asia/Shanghai',
targetSelector: { mode: 'fixed', instanceIds: ['a'] },
sms: { recipients: ['13800138000', '13900139000'], content: 'Maintenance' },
misfirePolicy: 'skip',
overlapPolicy: 'skip',
retryPolicy: { maxRetries: 1, intervalSeconds: 30 },
enabled: true,
},
smsSecretReference: 'memory://sms',
createdBy: 'operator',
now: '2026-07-30T00:00:00.000Z',
});
const attempts = new Map<string, number>();
const send = vi.fn(async (_instanceId: string, input: { phoneNumber: string }) => {
const count = (attempts.get(input.phoneNumber) ?? 0) + 1;
attempts.set(input.phoneNumber, count);
if (input.phoneNumber === '13800138000' && count === 1) throw new Error('temporary');
return { sent: true as const };
});
const sleep = vi.fn(async () => undefined);
const dispatcher = new ScheduledOperationDispatcher({
db,
operations: { prepare: vi.fn(), execute: vi.fn() },
messages: { send },
store: {
set: vi.fn(),
get: vi.fn(async () =>
JSON.stringify({ recipients: ['13800138000', '13900139000'], content: 'Maintenance' }),
),
delete: vi.fn(),
},
repository,
sleep,
});
const result = await dispatcher.dispatch(task, [{ id: 'a', revision: 1 }], {
actor: 'operator',
requestId: 'request-1',
});
expect(result.outcome).toBe('succeeded');
expect(send).toHaveBeenCalledTimes(3);
expect(sleep).toHaveBeenCalledTimes(1);
expect(sleep).toHaveBeenCalledWith(30_000);
db.close();
});
});
@@ -0,0 +1,240 @@
import { createHash, randomUUID } from 'node:crypto';
import type { Job, PrepareOperationRequest, ScheduledTask } from '@multi-simadmin/contracts';
import type { InstanceMessageService } from '../messages/instance-message-service.js';
import type { SecureOperationExecution } from '../operations/secure-operation-execution.js';
import type { SqliteDatabase } from '../../infrastructure/database/database.js';
import type { SecretStore } from '../../infrastructure/secrets/secret-store.js';
import type { DispatchResult } from './scheduler-coordinator.js';
import { ScheduledTaskRepository } from './scheduled-task-repository.js';
import type { ResolvedTarget } from './target-resolver.js';
interface OperationExecutor {
prepare(
input: PrepareOperationRequest,
requestId?: string,
): Promise<{ id: string; confirmationToken: string }>;
execute(
input: { preparationId: string; confirmationToken: string },
actor: string,
requestId: string,
): Promise<Pick<Job, 'id' | 'status'>>;
}
interface MessageExecutor {
send(
instanceId: string,
input: { phoneNumber: string; content: string },
): Promise<{ readonly sent: true }>;
}
export interface ScheduledOperationDispatcherOptions {
readonly db: SqliteDatabase;
readonly operations: Pick<SecureOperationExecution, 'prepare' | 'execute'> | OperationExecutor;
readonly messages: Pick<InstanceMessageService, 'send'> | MessageExecutor;
readonly store: SecretStore;
readonly repository: ScheduledTaskRepository;
readonly now?: () => Date;
readonly id?: () => string;
readonly sleep?: (milliseconds: number) => Promise<void>;
}
function aggregate(successes: number, failures: number): DispatchResult['outcome'] {
if (successes > 0 && failures > 0) return 'partially-succeeded';
return failures > 0 ? 'failed' : 'succeeded';
}
export class ScheduledOperationDispatcher {
private readonly now: () => Date;
private readonly id: () => string;
private readonly sleep: (milliseconds: number) => Promise<void>;
constructor(private readonly options: ScheduledOperationDispatcherOptions) {
this.now = options.now ?? (() => new Date());
this.id = options.id ?? randomUUID;
this.sleep =
options.sleep ??
((milliseconds) => new Promise((resolve) => setTimeout(resolve, milliseconds)));
}
async dispatch(
task: ScheduledTask,
targets: readonly ResolvedTarget[],
context: { readonly actor: string; readonly requestId: string },
): Promise<DispatchResult> {
if (task.operationType === 'send-sms') return this.dispatchSms(task, targets, context);
const operationId =
task.operationType === 'restart-service' ? 'postServiceRestart' : 'postSystemReboot';
const schema =
task.operationType === 'restart-service'
? 'simadmin.58e2204.postServiceRestart.parameters.v1'
: 'simadmin.58e2204.postSystemReboot.parameters.v1';
const fields =
task.operationType === 'reboot-system'
? [{ fieldId: 'delay_seconds', kind: 'number' as const, value: 3 }]
: [];
const jobIds: string[] = [];
let successes = 0;
let failures = 0;
for (const target of targets) {
let succeeded = false;
for (let attempt = 0; attempt <= task.retryPolicy.maxRetries; attempt += 1) {
try {
const preparation = await this.options.operations.prepare(
{
operationId,
targets: [{ instanceId: target.id, revision: target.revision }],
parameters: { parameterSchemaId: schema, fields },
},
context.requestId,
);
const job = await this.options.operations.execute(
{ preparationId: preparation.id, confirmationToken: preparation.confirmationToken },
context.actor,
context.requestId,
);
jobIds.push(job.id);
if (job.status === 'succeeded') {
succeeded = true;
break;
}
} catch {
// Continue only within the task's explicit bounded retry policy.
}
if (!succeeded && attempt < task.retryPolicy.maxRetries)
await this.sleep(task.retryPolicy.intervalSeconds * 1_000);
}
if (succeeded) successes += 1;
else failures += 1;
}
return { outcome: aggregate(successes, failures), jobIds };
}
private async dispatchSms(
task: ScheduledTask,
targets: readonly ResolvedTarget[],
context: { readonly actor: string; readonly requestId: string },
): Promise<DispatchResult> {
const reference = this.options.repository.getSmsSecretReference(task.id);
if (!reference)
return { outcome: 'needs-attention', reason: 'SMS secret is missing', jobIds: [] };
const secret = await this.options.store.get(reference);
if (!secret) return { outcome: 'needs-attention', reason: 'SMS secret is missing', jobIds: [] };
let payload: { recipients: string[]; content: string };
try {
payload = JSON.parse(secret) as typeof payload;
if (
!Array.isArray(payload.recipients) ||
!payload.recipients.length ||
typeof payload.content !== 'string' ||
!payload.content
)
throw new Error('invalid');
} catch {
return { outcome: 'needs-attention', reason: 'SMS secret is invalid', jobIds: [] };
}
const jobIds: string[] = [];
let successes = 0;
let failures = 0;
for (const target of targets) {
const result = await this.recordSmsJob(target.id, payload, task.retryPolicy, context);
jobIds.push(result.id);
if (result.succeeded) successes += 1;
else failures += 1;
}
return { outcome: aggregate(successes, failures), jobIds };
}
private async recordSmsJob(
instanceId: string,
payload: { recipients: readonly string[]; content: string },
retryPolicy: ScheduledTask['retryPolicy'],
context: { readonly actor: string; readonly requestId: string },
): Promise<{ readonly id: string; readonly succeeded: boolean }> {
const ids = { job: this.id(), item: this.id(), attempt: this.id(), audit: this.id() };
const startedAt = this.now().toISOString();
const digest = createHash('sha256').update(JSON.stringify(payload)).digest('hex');
this.options.db.transaction(() => {
this.options.db
.prepare(
`INSERT INTO jobs (id,root_job_id,operation_id,risk_level,status,requested_by,request_id,parameters_digest,created_at,started_at,updated_at)
VALUES (?,?,'sendSms','R2','running',?,?,?,?,?,?)`,
)
.run(
ids.job,
ids.job,
context.actor,
context.requestId,
digest,
startedAt,
startedAt,
startedAt,
);
this.options.db
.prepare(
"INSERT INTO job_attempts (id,job_id,status,started_at,created_at) VALUES (?,?,'running',?,?)",
)
.run(ids.attempt, ids.job, startedAt, startedAt);
this.options.db
.prepare(
"INSERT INTO job_items (id,job_id,instance_id,attempt_number,status,created_at,started_at,updated_at) VALUES (?,?,?,1,'running',?,?,?)",
)
.run(ids.item, ids.job, instanceId, startedAt, startedAt, startedAt);
})();
let succeeded = true;
for (const recipient of payload.recipients) {
let delivered = false;
for (let attempt = 0; attempt <= retryPolicy.maxRetries; attempt += 1) {
try {
await this.options.messages.send(instanceId, {
phoneNumber: recipient,
content: payload.content,
});
delivered = true;
break;
} catch {
delivered = false;
}
if (!delivered && attempt < retryPolicy.maxRetries)
await this.sleep(retryPolicy.intervalSeconds * 1_000);
}
if (!delivered) succeeded = false;
}
const finishedAt = this.now().toISOString();
const status = succeeded ? 'succeeded' : 'failed';
this.options.db.transaction(() => {
this.options.db
.prepare('UPDATE jobs SET status=?,finished_at=?,updated_at=? WHERE id=?')
.run(status, finishedAt, finishedAt, ids.job);
this.options.db
.prepare(
'UPDATE job_items SET status=?,result_code=?,finished_at=?,updated_at=? WHERE id=?',
)
.run(status, succeeded ? 'SMS_SENT' : 'SMS_FAILED', finishedAt, finishedAt, ids.item);
this.options.db
.prepare('UPDATE job_attempts SET status=?,finished_at=? WHERE id=?')
.run(status, finishedAt, ids.attempt);
this.options.db
.prepare(
`INSERT INTO audit_events
(id,instance_id,job_id,actor,operation_id,risk_level,request_id,parameters_summary_json,body_digest,result_code,duration_ms,created_at)
VALUES (?,?,?,?,'sendSms','R2',?,?,?,?,0,?)`,
)
.run(
ids.audit,
instanceId,
ids.job,
context.actor,
context.requestId,
JSON.stringify([
{ fieldId: 'recipients', displayValue: '[REDACTED]', redacted: true },
{ fieldId: 'content', displayValue: '[REDACTED]', redacted: true },
]),
digest,
succeeded ? 'SMS_SENT' : 'SMS_FAILED',
finishedAt,
);
})();
return { id: ids.job, succeeded };
}
}
@@ -0,0 +1,173 @@
import Database from 'better-sqlite3';
import { describe, expect, it } from 'vitest';
import { migrateDatabase } from '../../infrastructure/database/migrations.js';
import { ScheduledTaskRepository } from './scheduled-task-repository.js';
function fixture() {
const db = new Database(':memory:');
db.pragma('foreign_keys = ON');
migrateDatabase(db);
return { db, repository: new ScheduledTaskRepository(db) };
}
const task = {
name: 'Morning restart',
operationType: 'restart-service' as const,
cronExpression: '0 9 * * *',
timezone: 'Asia/Shanghai' as const,
targetSelector: { mode: 'fixed' as const, instanceIds: ['instance-a', 'instance-b'] },
misfirePolicy: 'skip' as const,
overlapPolicy: 'skip' as const,
retryPolicy: { maxRetries: 0, intervalSeconds: 60 },
enabled: true,
};
describe('ScheduledTaskRepository', () => {
it('round-trips a versioned task without exposing secret payload data', () => {
const { db, repository } = fixture();
const created = repository.create({
id: 'task-1',
task,
createdBy: 'operator',
now: '2026-07-30T00:00:00.000Z',
nextDueAt: '2026-07-30T01:00:00.000Z',
});
expect(created).toMatchObject({
id: 'task-1',
version: 1,
nextDueAt: '2026-07-30T01:00:00.000Z',
});
expect(repository.get('task-1')).toEqual(created);
expect(JSON.stringify(created)).not.toMatch(/secretReference|content|recipient/i);
db.close();
});
it('claims the same scheduled occurrence once and preserves its immutable snapshot', () => {
const { db, repository } = fixture();
repository.create({
id: 'task-1',
task,
createdBy: 'operator',
now: '2026-07-30T00:00:00.000Z',
});
const dueAt = '2026-07-30T01:00:00.000Z';
const first = repository.claimOccurrence({
id: 'run-1',
scheduledTaskId: 'task-1',
scheduleVersion: 1,
dueAt,
claimedAt: '2026-07-30T01:00:01.000Z',
triggerSource: 'scheduled',
targetSnapshot: ['instance-a', 'instance-b'],
taskSnapshot: task,
});
const duplicate = repository.claimOccurrence({
id: 'run-2',
scheduledTaskId: 'task-1',
scheduleVersion: 1,
dueAt,
claimedAt: '2026-07-30T01:00:02.000Z',
triggerSource: 'scheduled',
targetSnapshot: ['changed'],
taskSnapshot: task,
});
expect(first).toMatchObject({ id: 'run-1', targetSnapshot: ['instance-a', 'instance-b'] });
expect(duplicate).toBeNull();
expect(repository.getRun('run-1')?.targetSnapshot).toEqual(['instance-a', 'instance-b']);
db.close();
});
it('rolls back due-time advancement when scheduled claim preparation fails', () => {
const { db, repository } = fixture();
repository.create({
id: 'task-1',
task,
createdBy: 'operator',
now: '2026-07-30T00:00:00.000Z',
nextDueAt: '2026-07-30T01:00:00.000Z',
});
expect(() =>
repository.claimScheduledOccurrence({
id: 'run-1',
scheduledTaskId: 'task-1',
scheduleVersion: 1,
dueAt: '2026-07-30T01:00:00.000Z',
nextDueAt: '2026-07-31T01:00:00.000Z',
claimedAt: '2026-07-30T01:00:00.000Z',
taskSnapshot: task,
overlapPolicy: 'skip',
resolveTargets: () => {
throw new Error('target resolution failed');
},
}),
).toThrow('target resolution failed');
expect(repository.get('task-1')?.nextDueAt).toBe('2026-07-30T01:00:00.000Z');
expect(repository.listRuns()).toHaveLength(0);
db.close();
});
it('soft deletes task configuration while retaining completed run history', () => {
const { db, repository } = fixture();
repository.create({
id: 'task-1',
task,
createdBy: 'operator',
now: '2026-07-30T00:00:00.000Z',
});
repository.claimOccurrence({
id: 'run-1',
scheduledTaskId: 'task-1',
scheduleVersion: 1,
dueAt: '2026-07-30T01:00:00.000Z',
claimedAt: '2026-07-30T01:00:01.000Z',
triggerSource: 'scheduled',
targetSnapshot: [],
taskSnapshot: task,
});
repository.finishRun('run-1', {
outcome: 'no-targets',
reason: 'No enabled instances matched',
jobIds: [],
finishedAt: '2026-07-30T01:00:02.000Z',
});
repository.softDelete('task-1', 1, 'operator', '2026-07-30T02:00:00.000Z');
expect(repository.get('task-1')).toBeNull();
expect(repository.getRun('run-1')).toMatchObject({ outcome: 'no-targets' });
db.close();
});
it('reconciles interrupted started runs to needs-attention', () => {
const { db, repository } = fixture();
repository.create({
id: 'task-1',
task,
createdBy: 'operator',
now: '2026-07-30T00:00:00.000Z',
});
repository.claimOccurrence({
id: 'run-1',
scheduledTaskId: 'task-1',
scheduleVersion: 1,
dueAt: '2026-07-30T01:00:00.000Z',
claimedAt: '2026-07-30T01:00:00.000Z',
triggerSource: 'scheduled',
targetSnapshot: ['instance-a'],
taskSnapshot: task,
});
repository.startRun('run-1', '2026-07-30T01:00:01.000Z');
expect(repository.reconcileInterruptedRuns('2026-07-30T02:00:00.000Z')).toBe(1);
expect(repository.getRun('run-1')).toMatchObject({
outcome: 'needs-attention',
reason: 'Scheduler stopped before the run outcome was known',
finishedAt: '2026-07-30T02:00:00.000Z',
});
expect(repository.hasActiveRun('task-1')).toBe(false);
db.close();
});
});
@@ -0,0 +1,485 @@
import type {
CreateScheduledTaskRequest,
ScheduledRun,
ScheduledRunOutcome,
ScheduledRunTriggerSource,
ScheduledTask,
} from '@multi-simadmin/contracts';
import type { SqliteDatabase } from '../../infrastructure/database/database.js';
interface ScheduledTaskRow {
id: string;
name: string;
operation_type: ScheduledTask['operationType'];
enabled: number;
version: number;
cron_expression: string;
timezone: ScheduledTask['timezone'];
target_selector_json: string;
sms_secret_reference: string | null;
sms_recipient_count: number | null;
effective_start_at: string | null;
effective_end_at: string | null;
misfire_policy: ScheduledTask['misfirePolicy'];
overlap_policy: ScheduledTask['overlapPolicy'];
retry_policy_json: string;
next_due_at: string | null;
last_evaluated_at: string | null;
created_by: string;
updated_by: string;
created_at: string;
updated_at: string;
}
interface ScheduledRunRow {
id: string;
scheduled_task_id: string;
schedule_version: number;
task_snapshot_json: string;
due_at: string;
claimed_at: string;
started_at: string | null;
finished_at: string | null;
target_snapshot_json: string;
outcome: ScheduledRunOutcome | null;
reason: string | null;
job_ids_json: string;
trigger_source: ScheduledRunTriggerSource;
attempt: number;
}
export interface CreateTaskRecord {
readonly id: string;
readonly task: CreateScheduledTaskRequest;
readonly smsSecretReference?: string;
readonly createdBy: string;
readonly now: string;
readonly nextDueAt?: string;
}
export interface UpdateTaskRecord {
readonly id: string;
readonly version: number;
readonly task: CreateScheduledTaskRequest;
readonly smsSecretReference?: string;
readonly updatedBy: string;
readonly now: string;
readonly nextDueAt?: string;
}
export interface ClaimOccurrenceRecord {
readonly id: string;
readonly scheduledTaskId: string;
readonly scheduleVersion: number;
readonly dueAt: string;
readonly claimedAt: string;
readonly triggerSource: ScheduledRunTriggerSource;
readonly targetSnapshot: readonly string[];
readonly taskSnapshot: CreateScheduledTaskRequest;
readonly attempt?: number;
}
export interface ScheduledClaimTarget {
readonly id: string;
readonly revision: number;
}
export interface ClaimScheduledOccurrenceRecord {
readonly id: string;
readonly scheduledTaskId: string;
readonly scheduleVersion: number;
readonly dueAt: string;
readonly nextDueAt: string;
readonly claimedAt: string;
readonly taskSnapshot: CreateScheduledTaskRequest;
readonly overlapPolicy: ScheduledTask['overlapPolicy'];
readonly resolveTargets: () => {
readonly targets: readonly ScheduledClaimTarget[];
readonly targetSnapshot?: readonly string[];
readonly attentionReason?: string;
};
readonly terminal?: {
readonly outcome: ScheduledRunOutcome;
readonly reason: string;
};
}
export interface ScheduledClaimResult {
readonly run: ScheduledRun;
readonly targets: readonly ScheduledClaimTarget[];
readonly disposition: 'execute' | 'queued' | 'finished';
}
export interface FinishRunRecord {
readonly outcome: ScheduledRunOutcome;
readonly reason?: string;
readonly jobIds: readonly string[];
readonly finishedAt: string;
}
function parseJson<T>(value: string, label: string): T {
try {
return JSON.parse(value) as T;
} catch {
throw new Error(`Stored ${label} is invalid`);
}
}
function projectTask(row: ScheduledTaskRow): ScheduledTask {
return {
id: row.id,
name: row.name,
operationType: row.operation_type,
cronExpression: row.cron_expression,
timezone: row.timezone,
targetSelector: parseJson(row.target_selector_json, 'target selector'),
...(row.sms_secret_reference
? { sms: { configured: true, recipientCount: row.sms_recipient_count ?? 0 } }
: {}),
...(row.effective_start_at ? { effectiveStartAt: row.effective_start_at } : {}),
...(row.effective_end_at ? { effectiveEndAt: row.effective_end_at } : {}),
misfirePolicy: row.misfire_policy,
overlapPolicy: row.overlap_policy,
retryPolicy: parseJson(row.retry_policy_json, 'retry policy'),
enabled: row.enabled === 1,
version: row.version,
...(row.next_due_at ? { nextDueAt: row.next_due_at } : {}),
...(row.last_evaluated_at ? { lastEvaluatedAt: row.last_evaluated_at } : {}),
createdBy: row.created_by,
updatedBy: row.updated_by,
createdAt: row.created_at,
updatedAt: row.updated_at,
};
}
function projectRun(row: ScheduledRunRow): ScheduledRun {
const task = parseJson<CreateScheduledTaskRequest>(row.task_snapshot_json, 'task snapshot');
return {
id: row.id,
scheduledTaskId: row.scheduled_task_id,
scheduleVersion: row.schedule_version,
taskName: task.name,
operationType: task.operationType,
dueAt: row.due_at,
claimedAt: row.claimed_at,
...(row.started_at ? { startedAt: row.started_at } : {}),
...(row.finished_at ? { finishedAt: row.finished_at } : {}),
targetSnapshot: parseJson(row.target_snapshot_json, 'target snapshot'),
...(row.outcome ? { outcome: row.outcome } : {}),
...(row.reason ? { reason: row.reason } : {}),
jobIds: parseJson(row.job_ids_json, 'job ids'),
triggerSource: row.trigger_source,
attempt: row.attempt,
};
}
export class ScheduledTaskRepository {
constructor(private readonly db: SqliteDatabase) {}
create(input: CreateTaskRecord): ScheduledTask {
const task = input.task;
this.db
.prepare(
`INSERT INTO scheduled_tasks
(id,name,operation_type,enabled,version,cron_expression,timezone,target_selector_json,
sms_secret_reference,sms_recipient_count,effective_start_at,effective_end_at,misfire_policy,
overlap_policy,retry_policy_json,next_due_at,created_by,updated_by,created_at,updated_at)
VALUES (?,?,?,?,1,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`,
)
.run(
input.id,
task.name,
task.operationType,
task.enabled ? 1 : 0,
task.cronExpression,
task.timezone,
JSON.stringify(task.targetSelector),
input.smsSecretReference ?? null,
task.sms?.recipients.length ?? null,
task.effectiveStartAt ?? null,
task.effectiveEndAt ?? null,
task.misfirePolicy,
task.overlapPolicy,
JSON.stringify(task.retryPolicy),
input.nextDueAt ?? null,
input.createdBy,
input.createdBy,
input.now,
input.now,
);
const created = this.get(input.id);
if (!created) throw new Error('Scheduled task was not persisted');
return created;
}
update(input: UpdateTaskRecord): ScheduledTask {
const task = input.task;
const result = this.db
.prepare(
`UPDATE scheduled_tasks
SET name = ?, operation_type = ?, enabled = ?, version = version + 1,
cron_expression = ?, timezone = ?, target_selector_json = ?,
sms_secret_reference = ?, sms_recipient_count = ?, effective_start_at = ?,
effective_end_at = ?, misfire_policy = ?, overlap_policy = ?, retry_policy_json = ?,
next_due_at = ?, last_evaluated_at = NULL, updated_by = ?, updated_at = ?
WHERE id = ? AND version = ? AND deleted_at IS NULL`,
)
.run(
task.name,
task.operationType,
task.enabled ? 1 : 0,
task.cronExpression,
task.timezone,
JSON.stringify(task.targetSelector),
input.smsSecretReference ?? null,
task.sms?.recipients.length ?? null,
task.effectiveStartAt ?? null,
task.effectiveEndAt ?? null,
task.misfirePolicy,
task.overlapPolicy,
JSON.stringify(task.retryPolicy),
input.nextDueAt ?? null,
input.updatedBy,
input.now,
input.id,
input.version,
);
if (result.changes !== 1) throw new Error('Scheduled task was not found or version changed');
const updated = this.get(input.id);
if (!updated) throw new Error('Scheduled task was not persisted');
return updated;
}
get(id: string): ScheduledTask | null {
const row = this.db
.prepare('SELECT * FROM scheduled_tasks WHERE id = ? AND deleted_at IS NULL')
.get(id) as ScheduledTaskRow | undefined;
return row ? projectTask(row) : null;
}
list(): readonly ScheduledTask[] {
return (
this.db
.prepare(
'SELECT * FROM scheduled_tasks WHERE deleted_at IS NULL ORDER BY updated_at DESC, id ASC',
)
.all() as ScheduledTaskRow[]
).map(projectTask);
}
getSmsSecretReference(id: string): string | undefined {
const row = this.db
.prepare(
'SELECT sms_secret_reference FROM scheduled_tasks WHERE id = ? AND deleted_at IS NULL',
)
.get(id) as { sms_secret_reference: string | null } | undefined;
return row?.sms_secret_reference ?? undefined;
}
setEnabled(
id: string,
version: number,
enabled: boolean,
actor: string,
now: string,
): ScheduledTask {
const result = this.db
.prepare(
`UPDATE scheduled_tasks
SET enabled = ?, version = version + 1, updated_by = ?, updated_at = ?
WHERE id = ? AND version = ? AND deleted_at IS NULL`,
)
.run(enabled ? 1 : 0, actor, now, id, version);
if (result.changes !== 1) throw new Error('Scheduled task was not found or version changed');
const task = this.get(id);
if (!task) throw new Error('Scheduled task was not persisted');
return task;
}
advanceNextDue(id: string, version: number, nextDueAt: string, evaluatedAt: string): boolean {
return (
this.db
.prepare(
`UPDATE scheduled_tasks SET next_due_at = ?, last_evaluated_at = ?
WHERE id = ? AND version = ? AND enabled = 1 AND deleted_at IS NULL`,
)
.run(nextDueAt, evaluatedAt, id, version).changes === 1
);
}
startRun(id: string, startedAt: string): void {
const result = this.db
.prepare(
'UPDATE scheduled_runs SET started_at = ? WHERE id = ? AND started_at IS NULL AND finished_at IS NULL',
)
.run(startedAt, id);
if (result.changes !== 1) throw new Error('Scheduled run is missing or already started');
}
hasActiveRun(scheduledTaskId: string): boolean {
return !!this.db
.prepare(
'SELECT 1 FROM scheduled_runs WHERE scheduled_task_id = ? AND started_at IS NOT NULL AND finished_at IS NULL LIMIT 1',
)
.get(scheduledTaskId);
}
reconcileInterruptedRuns(finishedAt: string): number {
return this.db
.prepare(
`UPDATE scheduled_runs
SET outcome = 'needs-attention',
reason = 'Scheduler stopped before the run outcome was known',
finished_at = ?
WHERE started_at IS NOT NULL AND finished_at IS NULL`,
)
.run(finishedAt).changes;
}
getQueuedRun(scheduledTaskId: string): ScheduledRun | null {
const row = this.db
.prepare(
`SELECT * FROM scheduled_runs
WHERE scheduled_task_id = ? AND started_at IS NULL AND finished_at IS NULL
ORDER BY claimed_at ASC, id ASC LIMIT 1`,
)
.get(scheduledTaskId) as ScheduledRunRow | undefined;
return row ? projectRun(row) : null;
}
claimOccurrence(input: ClaimOccurrenceRecord): ScheduledRun | null {
const result = this.db
.prepare(
`INSERT OR IGNORE INTO scheduled_runs
(id,scheduled_task_id,schedule_version,task_snapshot_json,due_at,claimed_at,target_snapshot_json,
job_ids_json,trigger_source,attempt)
VALUES (?,?,?,?,?,?,?,?,?,?)`,
)
.run(
input.id,
input.scheduledTaskId,
input.scheduleVersion,
JSON.stringify(input.taskSnapshot),
input.dueAt,
input.claimedAt,
JSON.stringify(input.targetSnapshot),
'[]',
input.triggerSource,
input.attempt ?? 1,
);
return result.changes === 0 ? null : this.getRun(input.id);
}
claimScheduledOccurrence(input: ClaimScheduledOccurrenceRecord): ScheduledClaimResult | null {
return this.db.transaction((): ScheduledClaimResult | null => {
const advanced = this.db
.prepare(
`UPDATE scheduled_tasks SET next_due_at = ?, last_evaluated_at = ?
WHERE id = ? AND version = ? AND enabled = 1 AND deleted_at IS NULL AND next_due_at = ?`,
)
.run(
input.nextDueAt,
input.claimedAt,
input.scheduledTaskId,
input.scheduleVersion,
input.dueAt,
);
if (advanced.changes !== 1) return null;
const resolution = input.terminal
? { targets: [] as readonly ScheduledClaimTarget[] }
: input.resolveTargets();
const targets = [...resolution.targets];
let disposition: ScheduledClaimResult['disposition'] = 'execute';
let terminal =
input.terminal ??
(resolution.attentionReason
? { outcome: 'needs-attention' as const, reason: resolution.attentionReason }
: undefined);
if (!terminal && this.hasActiveRun(input.scheduledTaskId)) {
const queued = this.getQueuedRun(input.scheduledTaskId);
if (input.overlapPolicy === 'queue-once' && !queued) {
disposition = 'queued';
} else {
disposition = 'finished';
terminal = {
outcome: 'skipped',
reason:
input.overlapPolicy === 'queue-once'
? 'Overlap queue is full'
: 'Overlapping occurrence skipped by task policy',
};
}
}
const claimed = this.claimOccurrence({
id: input.id,
scheduledTaskId: input.scheduledTaskId,
scheduleVersion: input.scheduleVersion,
dueAt: input.dueAt,
claimedAt: input.claimedAt,
triggerSource: 'scheduled',
targetSnapshot: resolution.targetSnapshot ?? targets.map((target) => target.id),
taskSnapshot: input.taskSnapshot,
});
if (!claimed) return null;
if (terminal) {
return {
run: this.finishRun(claimed.id, {
outcome: terminal.outcome,
reason: terminal.reason,
jobIds: [],
finishedAt: input.claimedAt,
}),
targets,
disposition: 'finished',
};
}
return { run: claimed, targets, disposition };
})();
}
getRun(id: string): ScheduledRun | null {
const row = this.db.prepare('SELECT * FROM scheduled_runs WHERE id = ?').get(id) as
| ScheduledRunRow
| undefined;
return row ? projectRun(row) : null;
}
listRuns(scheduledTaskId?: string): readonly ScheduledRun[] {
const rows = scheduledTaskId
? (this.db
.prepare(
'SELECT * FROM scheduled_runs WHERE scheduled_task_id = ? ORDER BY due_at DESC, id ASC',
)
.all(scheduledTaskId) as ScheduledRunRow[])
: (this.db
.prepare('SELECT * FROM scheduled_runs ORDER BY due_at DESC, id ASC')
.all() as ScheduledRunRow[]);
return rows.map(projectRun);
}
finishRun(id: string, input: FinishRunRecord): ScheduledRun {
const result = this.db
.prepare(
`UPDATE scheduled_runs SET outcome = ?, reason = ?, job_ids_json = ?, finished_at = ?
WHERE id = ? AND finished_at IS NULL`,
)
.run(input.outcome, input.reason ?? null, JSON.stringify(input.jobIds), input.finishedAt, id);
if (result.changes !== 1) throw new Error('Scheduled run is missing or already finished');
const run = this.getRun(id);
if (!run) throw new Error('Scheduled run was not persisted');
return run;
}
softDelete(id: string, version: number, actor: string, now: string): void {
const result = this.db
.prepare(
`UPDATE scheduled_tasks SET enabled = 0, version = version + 1, updated_by = ?, updated_at = ?, deleted_at = ?
WHERE id = ? AND version = ? AND deleted_at IS NULL`,
)
.run(actor, now, now, id, version);
if (result.changes !== 1) throw new Error('Scheduled task was not found or version changed');
}
}
@@ -0,0 +1,235 @@
import Database from 'better-sqlite3';
import { describe, expect, it } from 'vitest';
import { migrateDatabase } from '../../infrastructure/database/migrations.js';
import type { SecretKey, SecretStore } from '../../infrastructure/secrets/secret-store.js';
import { ScheduledTaskRepository } from './scheduled-task-repository.js';
import { ScheduledTaskService } from './scheduled-task-service.js';
class MemorySecrets implements SecretStore {
readonly values = new Map<string, string>();
readonly setCalls: Array<{ key: SecretKey; value: string }> = [];
blockRotations = false;
private rotationWaiters: Array<() => void> = [];
async set(key: SecretKey, value: string): Promise<string> {
const reference = `memory://${key.instanceId}/${key.purpose}/${key.slot ?? 'default'}`;
this.setCalls.push({ key, value });
this.values.set(reference, value);
if (this.blockRotations && key.slot) {
await new Promise<void>((resolve) => {
this.rotationWaiters.push(resolve);
if (this.rotationWaiters.length === 2) {
for (const waiter of this.rotationWaiters.splice(0)) waiter();
}
});
}
return reference;
}
async get(reference: string): Promise<string | undefined> {
return this.values.get(reference);
}
async delete(reference: string): Promise<boolean> {
return this.values.delete(reference);
}
}
function fixture() {
const db = new Database(':memory:');
db.pragma('foreign_keys = ON');
migrateDatabase(db);
const store = new MemorySecrets();
const repository = new ScheduledTaskRepository(db);
let id = 0;
const service = new ScheduledTaskService({
repository,
store,
now: () => new Date('2026-07-30T00:00:00.000Z'),
id: () => `task-${++id}`,
});
return { db, repository, service, store };
}
const base = {
name: 'Morning SMS',
operationType: 'send-sms',
cronExpression: '0 9 * * *',
targetSelector: { mode: 'tags', match: 'all', tags: ['lab'] },
sms: { recipients: ['13800138000', '13900139000'], content: 'Maintenance complete' },
};
describe('ScheduledTaskService', () => {
it('stores SMS payload only in the secret store and returns a redacted summary', async () => {
const { db, service, store } = fixture();
const created = await service.create('operator', base);
expect(created.sms).toEqual({ configured: true, recipientCount: 2 });
expect(JSON.stringify(created)).not.toContain('13800138000');
expect(JSON.stringify(created)).not.toContain('Maintenance complete');
expect(store.setCalls[0]?.value).toBe(
JSON.stringify({
recipients: ['13800138000', '13900139000'],
content: 'Maintenance complete',
}),
);
const persisted = db.prepare('SELECT * FROM scheduled_tasks WHERE id = ?').get('task-1');
expect(JSON.stringify(persisted)).not.toContain('13800138000');
expect(JSON.stringify(persisted)).not.toContain('Maintenance complete');
db.close();
});
it('increments the optimistic version when pausing and rejects a stale update', async () => {
const { db, service } = fixture();
const created = await service.create('operator', {
...base,
operationType: 'restart-service',
sms: undefined,
});
expect(service.setEnabled('operator', created.id, created.version, false).version).toBe(2);
expect(() => service.setEnabled('operator', created.id, created.version, true)).toThrow(
/version/i,
);
db.close();
});
it('does not schedule the first occurrence before the effective start', async () => {
const { db, service } = fixture();
const created = await service.create('operator', {
...base,
operationType: 'restart-service',
sms: undefined,
effectiveStartAt: '2026-08-05T00:00:00.000Z',
});
expect(created.nextDueAt).toBe('2026-08-05T01:00:00.000Z');
db.close();
});
it('keeps immutable run history after deleting its schedule', async () => {
const { db, repository, service } = fixture();
const created = await service.create('operator', {
...base,
operationType: 'restart-service',
sms: undefined,
});
repository.claimOccurrence({
id: 'run-1',
scheduledTaskId: created.id,
scheduleVersion: created.version,
dueAt: '2026-07-30T01:00:00.000Z',
claimedAt: '2026-07-30T01:00:00.000Z',
triggerSource: 'manual',
targetSnapshot: [],
taskSnapshot: {
name: base.name,
cronExpression: base.cronExpression,
targetSelector: { mode: 'tags' as const, match: 'all' as const, tags: ['lab'] },
operationType: 'restart-service',
timezone: 'Asia/Shanghai',
misfirePolicy: 'skip',
overlapPolicy: 'skip',
retryPolicy: { maxRetries: 0, intervalSeconds: 60 },
enabled: true,
},
});
await service.remove('operator', created.id, created.version);
expect(service.get(created.id)).toBeNull();
expect(repository.getRun('run-1')).not.toBeNull();
db.close();
});
it('updates the complete task, preserves an unchanged SMS secret, and advances the version', async () => {
const { db, service, store } = fixture();
const created = await service.create('operator', base);
const reference = store.setCalls[0]?.key;
const updated = await service.update('editor', created.id, created.version, {
name: 'Evening SMS',
cronExpression: '30 18 * * *',
effectiveStartAt: '2026-08-01T00:00:00.000Z',
effectiveEndAt: '2026-09-01T00:00:00.000Z',
overlapPolicy: 'queue-once',
});
expect(updated).toMatchObject({
name: 'Evening SMS',
cronExpression: '30 18 * * *',
overlapPolicy: 'queue-once',
version: 2,
updatedBy: 'editor',
sms: { configured: true, recipientCount: 2 },
});
expect(store.setCalls).toHaveLength(1);
expect(store.setCalls[0]?.key).toEqual(reference);
db.close();
});
it('rotates edited SMS secrets and validates retry policy against the effective operation', async () => {
const { db, service, store } = fixture();
const created = await service.create('operator', base);
const oldReference = [...store.values.keys()][0];
const updated = await service.update('editor', created.id, created.version, {
sms: { recipients: ['13700137000'], content: 'Updated content' },
});
expect(updated.sms).toEqual({ configured: true, recipientCount: 1 });
expect(store.setCalls).toHaveLength(2);
expect(oldReference ? store.values.has(oldReference) : true).toBe(false);
const reboot = await service.create('operator', {
...base,
name: 'Reboot',
operationType: 'reboot-system',
sms: undefined,
});
await expect(
service.update('editor', reboot.id, reboot.version, {
retryPolicy: { maxRetries: 1, intervalSeconds: 60 },
}),
).rejects.toThrow(/reboot/i);
db.close();
});
it('does not delete the winning SMS secret when concurrent updates race', async () => {
const { db, repository, service, store } = fixture();
const created = await service.create('operator', base);
store.blockRotations = true;
const results = await Promise.allSettled([
service.update('editor-a', created.id, created.version, {
sms: { recipients: ['13700137000'], content: 'Update A' },
}),
service.update('editor-b', created.id, created.version, {
sms: { recipients: ['13600136000'], content: 'Update B' },
}),
]);
expect(results.filter((result) => result.status === 'fulfilled')).toHaveLength(1);
expect(results.filter((result) => result.status === 'rejected')).toHaveLength(1);
const currentReference = repository.getSmsSecretReference(created.id);
expect(currentReference).toBeDefined();
expect(currentReference ? store.values.has(currentReference) : false).toBe(true);
expect(store.values).toHaveLength(1);
db.close();
});
it('duplicates configuration and SMS secrets into a disabled independent task', async () => {
const { db, service, store } = fixture();
const created = await service.create('operator', base);
const duplicate = await service.duplicate('operator', created.id, created.version);
expect(duplicate).toMatchObject({
id: 'task-2',
name: 'Morning SMS copy',
operationType: 'send-sms',
enabled: false,
version: 1,
sms: { configured: true, recipientCount: 2 },
});
expect(store.setCalls).toHaveLength(2);
expect(store.setCalls[1]?.value).toBe(store.setCalls[0]?.value);
db.close();
});
});
@@ -0,0 +1,220 @@
import { randomUUID } from 'node:crypto';
import {
parseCreateScheduledTaskRequest,
parseUpdateScheduledTaskRequest,
type CreateScheduledTaskRequest,
type ScheduledTask,
type ScheduledSmsInput,
} from '@multi-simadmin/contracts';
import type { SecretStore } from '../../infrastructure/secrets/secret-store.js';
import { nextOccurrence, previewCron } from './schedule-time.js';
import { ScheduledTaskRepository } from './scheduled-task-repository.js';
export interface ScheduledTaskServiceOptions {
readonly repository: ScheduledTaskRepository;
readonly store: SecretStore;
readonly now?: () => Date;
readonly id?: () => string;
readonly secretSlot?: () => string;
}
export class ScheduledTaskService {
private readonly repository: ScheduledTaskRepository;
private readonly store: SecretStore;
private readonly now: () => Date;
private readonly id: () => string;
private readonly secretSlot: () => string;
constructor(options: ScheduledTaskServiceOptions) {
this.repository = options.repository;
this.store = options.store;
this.now = options.now ?? (() => new Date());
this.id = options.id ?? randomUUID;
this.secretSlot = options.secretSlot ?? randomUUID;
}
async create(actor: string, value: unknown): Promise<ScheduledTask> {
const task = parseCreateScheduledTaskRequest(value);
const now = this.now();
previewCron(task.cronExpression, 1, now);
const id = this.id();
let smsSecretReference: string | undefined;
if (task.sms) {
smsSecretReference = await this.store.set(
{ instanceId: id, purpose: 'scheduled-sms' },
JSON.stringify(task.sms),
);
}
try {
return this.repository.create({
id,
task,
...(smsSecretReference ? { smsSecretReference } : {}),
createdBy: actor,
now: now.toISOString(),
nextDueAt: this.nextDueAt(task, now),
});
} catch (error) {
if (smsSecretReference) await this.store.delete(smsSecretReference).catch(() => false);
throw error;
}
}
async update(actor: string, id: string, version: number, value: unknown): Promise<ScheduledTask> {
const source = this.updateObject(value);
const change = parseUpdateScheduledTaskRequest({ ...source, version });
const current = this.repository.get(id);
if (!current || current.version !== version)
throw new Error('Scheduled task was not found or version changed');
const oldReference = this.repository.getSmsSecretReference(id);
const operationType = change.operationType ?? current.operationType;
let smsPayload: ScheduledSmsInput | undefined;
if (operationType === 'send-sms') {
if (change.sms === null) throw new TypeError('SMS configuration is required');
if (change.sms) smsPayload = change.sms;
else smsPayload = await this.readSms(oldReference);
} else if (change.sms) {
throw new TypeError('SMS is only valid for send-sms');
}
const merged = parseCreateScheduledTaskRequest({
name: change.name ?? current.name,
operationType,
cronExpression: change.cronExpression ?? current.cronExpression,
timezone: change.timezone ?? current.timezone,
targetSelector: change.targetSelector ?? current.targetSelector,
...(smsPayload ? { sms: smsPayload } : {}),
...(change.effectiveStartAt === null
? {}
: change.effectiveStartAt
? { effectiveStartAt: change.effectiveStartAt }
: current.effectiveStartAt
? { effectiveStartAt: current.effectiveStartAt }
: {}),
...(change.effectiveEndAt === null
? {}
: change.effectiveEndAt
? { effectiveEndAt: change.effectiveEndAt }
: current.effectiveEndAt
? { effectiveEndAt: current.effectiveEndAt }
: {}),
misfirePolicy: change.misfirePolicy ?? current.misfirePolicy,
overlapPolicy: change.overlapPolicy ?? current.overlapPolicy,
retryPolicy: change.retryPolicy ?? current.retryPolicy,
enabled: change.enabled ?? current.enabled,
});
const now = this.now();
previewCron(merged.cronExpression, 1, now);
let nextReference = operationType === 'send-sms' ? oldReference : undefined;
let wroteReference: string | undefined;
if (operationType === 'send-sms' && change.sms) {
wroteReference = await this.store.set(
{ instanceId: id, purpose: 'scheduled-sms', slot: this.secretSlot() },
JSON.stringify(change.sms),
);
nextReference = wroteReference;
}
if (operationType === 'send-sms' && !nextReference)
throw new Error('Scheduled SMS secret is missing');
try {
const updated = this.repository.update({
id,
version,
task: merged,
...(nextReference ? { smsSecretReference: nextReference } : {}),
updatedBy: actor,
now: now.toISOString(),
nextDueAt: this.nextDueAt(merged, now),
});
if (oldReference && oldReference !== nextReference)
await this.store.delete(oldReference).catch(() => false);
return updated;
} catch (error) {
if (wroteReference) await this.store.delete(wroteReference).catch(() => false);
throw error;
}
}
async duplicate(actor: string, id: string, version: number): Promise<ScheduledTask> {
const current = this.repository.get(id);
if (!current || current.version !== version)
throw new Error('Scheduled task was not found or version changed');
const smsPayload =
current.operationType === 'send-sms'
? await this.readSms(this.repository.getSmsSecretReference(id))
: undefined;
const suffix = ' copy';
return this.create(actor, {
name: `${current.name.slice(0, 120 - suffix.length)}${suffix}`,
operationType: current.operationType,
cronExpression: current.cronExpression,
timezone: current.timezone,
targetSelector: current.targetSelector,
...(smsPayload ? { sms: smsPayload } : {}),
...(current.effectiveStartAt ? { effectiveStartAt: current.effectiveStartAt } : {}),
...(current.effectiveEndAt ? { effectiveEndAt: current.effectiveEndAt } : {}),
misfirePolicy: current.misfirePolicy,
overlapPolicy: current.overlapPolicy,
retryPolicy: current.retryPolicy,
enabled: false,
});
}
get(id: string): ScheduledTask | null {
return this.repository.get(id);
}
list(): readonly ScheduledTask[] {
return this.repository.list();
}
preview(cronExpression: string, count = 5, from = this.now()): readonly string[] {
return previewCron(cronExpression, count, from);
}
setEnabled(actor: string, id: string, version: number, enabled: boolean): ScheduledTask {
return this.repository.setEnabled(id, version, enabled, actor, this.now().toISOString());
}
async remove(actor: string, id: string, version: number): Promise<void> {
const reference = this.repository.getSmsSecretReference(id);
this.repository.softDelete(id, version, actor, this.now().toISOString());
if (reference) await this.store.delete(reference).catch(() => false);
}
private updateObject(value: unknown): Record<string, unknown> {
if (typeof value !== 'object' || value === null || Array.isArray(value))
throw new TypeError('Schedule update must be an object');
return value as Record<string, unknown>;
}
private async readSms(reference: string | undefined): Promise<ScheduledSmsInput> {
if (!reference) throw new Error('Scheduled SMS secret is missing');
const value = await this.store.get(reference);
if (!value) throw new Error('Scheduled SMS secret is missing');
try {
const parsed = JSON.parse(value) as unknown;
return parseCreateScheduledTaskRequest({
name: 'SMS validation',
operationType: 'send-sms',
cronExpression: '0 0 * * *',
timezone: 'Asia/Shanghai',
targetSelector: { mode: 'fixed', instanceIds: ['secret-validation'] },
sms: parsed,
}).sms as ScheduledSmsInput;
} catch {
throw new Error('Scheduled SMS secret is invalid');
}
}
private nextDueAt(task: CreateScheduledTaskRequest, now: Date): string {
const anchor = task.effectiveStartAt
? new Date(Math.max(now.getTime(), Date.parse(task.effectiveStartAt)))
: now;
return nextOccurrence(task.cronExpression, anchor);
}
}
@@ -0,0 +1,321 @@
import Database from 'better-sqlite3';
import { describe, expect, it, vi } from 'vitest';
import { migrateDatabase } from '../../infrastructure/database/migrations.js';
import { ScheduledTaskRepository } from './scheduled-task-repository.js';
import { SchedulerCoordinator } from './scheduler-coordinator.js';
import type { ScheduledTask } from '@multi-simadmin/contracts';
import type { ResolvedTarget } from './target-resolver.js';
function fixture() {
const db = new Database(':memory:');
db.pragma('foreign_keys = ON');
migrateDatabase(db);
const now = '2026-07-30T01:00:00.000Z';
db.prepare(
'INSERT INTO instances (id,name,base_url,enabled,config_revision,created_at,updated_at) VALUES (?,?,?,?,?,?,?)',
).run('a', 'Alpha', 'http://10.0.0.1', 1, 2, now, now);
db.prepare(
'INSERT INTO instances (id,name,base_url,enabled,config_revision,created_at,updated_at) VALUES (?,?,?,?,?,?,?)',
).run('b', 'Beta', 'http://10.0.0.2', 1, 3, now, now);
const capability = db.prepare(
"INSERT INTO capabilities (instance_id,operation_id,state,observed_at,created_at,updated_at) VALUES (?,'postServiceRestart','supported',?,?,?)",
);
capability.run('a', now, now, now);
capability.run('b', now, now, now);
const repository = new ScheduledTaskRepository(db);
repository.create({
id: 'task-1',
task: {
name: 'Restart lab',
operationType: 'restart-service',
cronExpression: '0 9 * * *',
timezone: 'Asia/Shanghai',
targetSelector: { mode: 'fixed', instanceIds: ['a', 'b'] },
misfirePolicy: 'skip',
overlapPolicy: 'skip',
retryPolicy: { maxRetries: 0, intervalSeconds: 60 },
enabled: true,
},
createdBy: 'operator',
now,
nextDueAt: now,
});
return { db, repository };
}
describe('SchedulerCoordinator', () => {
it('resolves multiple targets, persists the snapshot, and finishes a manual run', async () => {
const { db, repository } = fixture();
const dispatch = vi.fn<
(
task: ScheduledTask,
targets: readonly ResolvedTarget[],
) => Promise<{ outcome: 'succeeded'; jobIds: string[] }>
>(async () => ({
outcome: 'succeeded',
jobIds: ['job-a', 'job-b'],
}));
const coordinator = new SchedulerCoordinator({
db,
repository,
dispatch,
now: () => new Date('2026-07-30T01:00:00.000Z'),
id: () => 'run-1',
});
const run = await coordinator.runNow('task-1', 'operator', 'request-1');
expect(run).toMatchObject({
id: 'run-1',
triggerSource: 'manual',
targetSnapshot: ['a', 'b'],
outcome: 'succeeded',
jobIds: ['job-a', 'job-b'],
});
expect(dispatch.mock.calls[0]?.[1]).toEqual([
{ id: 'a', revision: 2 },
{ id: 'b', revision: 3 },
]);
db.close();
});
it('does not manually dispatch when a selected target lacks the required capability', async () => {
const { db, repository } = fixture();
db.prepare(
"UPDATE capabilities SET state = 'unsupported' WHERE instance_id = 'b' AND operation_id = 'postServiceRestart'",
).run();
const dispatch = vi.fn();
const coordinator = new SchedulerCoordinator({
db,
repository,
dispatch,
now: () => new Date('2026-07-30T01:00:00.000Z'),
id: () => 'manual-capability-run',
});
const run = await coordinator.runNow('task-1', 'operator', 'request-1');
expect(run).toMatchObject({ outcome: 'needs-attention', targetSnapshot: ['a', 'b'] });
expect(dispatch).not.toHaveBeenCalled();
db.close();
});
it('records no-targets without dispatching and prevents duplicate scheduled claims', async () => {
const { db, repository } = fixture();
db.prepare('UPDATE instances SET enabled = 0').run();
const dispatch = vi.fn();
const coordinator = new SchedulerCoordinator({
db,
repository,
dispatch,
now: () => new Date('2026-07-30T01:00:00.000Z'),
id: () => 'run-1',
});
const first = await coordinator.tick();
const second = await coordinator.tick();
expect(first).toHaveLength(1);
expect(first[0]).toMatchObject({ outcome: 'no-targets' });
expect(second).toHaveLength(0);
expect(dispatch).not.toHaveBeenCalled();
db.close();
});
it('records needs-attention when a matched target loses its required capability', async () => {
const { db, repository } = fixture();
db.prepare(
"UPDATE capabilities SET state = 'unsupported' WHERE instance_id = 'b' AND operation_id = 'postServiceRestart'",
).run();
const dispatch = vi.fn();
const coordinator = new SchedulerCoordinator({
db,
repository,
dispatch,
now: () => new Date('2026-07-30T01:00:00.000Z'),
id: () => 'capability-run',
});
const runs = await coordinator.tick();
expect(runs).toHaveLength(1);
expect(runs[0]).toMatchObject({
outcome: 'needs-attention',
targetSnapshot: ['a', 'b'],
});
expect(runs[0]?.reason).toContain('b');
expect(dispatch).not.toHaveBeenCalled();
db.close();
});
it('queues one overlapping occurrence, skips a second, and dispatches the queued run later', async () => {
const { db, repository } = fixture();
db.prepare("UPDATE scheduled_tasks SET overlap_policy = 'queue-once'").run();
const task = repository.get('task-1')!;
const taskSnapshot = {
name: task.name,
operationType: task.operationType,
cronExpression: task.cronExpression,
timezone: task.timezone,
targetSelector: task.targetSelector,
misfirePolicy: task.misfirePolicy,
overlapPolicy: task.overlapPolicy,
retryPolicy: task.retryPolicy,
enabled: task.enabled,
};
repository.claimOccurrence({
id: 'active-run',
scheduledTaskId: task.id,
scheduleVersion: task.version,
dueAt: '2026-07-30T00:00:00.000Z',
claimedAt: '2026-07-30T00:00:00.000Z',
triggerSource: 'manual',
targetSnapshot: ['a'],
taskSnapshot,
});
repository.startRun('active-run', '2026-07-30T00:00:00.000Z');
let now = new Date('2026-07-30T01:00:00.000Z');
let id = 0;
const dispatch = vi.fn(async () => ({ outcome: 'succeeded' as const, jobIds: ['job-1'] }));
const coordinator = new SchedulerCoordinator({
db,
repository,
dispatch,
now: () => now,
id: () => `queued-${++id}`,
});
expect(await coordinator.tick()).toHaveLength(0);
expect(repository.getQueuedRun('task-1')).toMatchObject({
id: 'queued-1',
targetSnapshot: ['a', 'b'],
});
now = new Date('2026-07-31T01:00:00.000Z');
const queueFull = await coordinator.tick();
expect(queueFull).toHaveLength(1);
expect(queueFull[0]).toMatchObject({ outcome: 'skipped', reason: 'Overlap queue is full' });
expect(dispatch).not.toHaveBeenCalled();
repository.finishRun('active-run', {
outcome: 'succeeded',
jobIds: [],
finishedAt: now.toISOString(),
});
const completed = await coordinator.tick();
expect(completed[0]).toMatchObject({ id: 'queued-1', outcome: 'succeeded' });
expect(dispatch).toHaveBeenCalledTimes(1);
db.close();
});
it('records and advances an occurrence outside the effective window', async () => {
const { db, repository } = fixture();
db.prepare("UPDATE scheduled_tasks SET effective_end_at = '2026-07-29T23:59:00.000Z'").run();
const dispatch = vi.fn();
const coordinator = new SchedulerCoordinator({
db,
repository,
dispatch,
now: () => new Date('2026-07-30T01:00:00.000Z'),
id: () => 'window-skip',
});
const runs = await coordinator.tick();
expect(runs).toHaveLength(1);
expect(runs[0]).toMatchObject({
outcome: 'skipped',
reason: 'Occurrence is outside the effective window',
});
expect(repository.get('task-1')?.nextDueAt).toBe('2026-07-31T01:00:00.000Z');
expect(dispatch).not.toHaveBeenCalled();
db.close();
});
it('does not advance a due occurrence when its claim cannot be persisted', async () => {
const { db, repository } = fixture();
vi.spyOn(repository, 'claimOccurrence').mockImplementationOnce(() => {
throw new Error('claim failed');
});
const coordinator = new SchedulerCoordinator({
db,
repository,
dispatch: vi.fn(),
now: () => new Date('2026-07-30T01:00:00.000Z'),
id: () => 'failed-claim',
});
await expect(coordinator.tick()).rejects.toThrow('claim failed');
expect(repository.get('task-1')?.nextDueAt).toBe('2026-07-30T01:00:00.000Z');
db.close();
});
it('executes at most the queued occurrence for a task during one tick', async () => {
const { db, repository } = fixture();
const task = repository.get('task-1')!;
repository.claimOccurrence({
id: 'queued-run',
scheduledTaskId: task.id,
scheduleVersion: task.version,
dueAt: '2026-07-29T01:00:00.000Z',
claimedAt: '2026-07-29T01:00:00.000Z',
triggerSource: 'scheduled',
targetSnapshot: ['a'],
taskSnapshot: {
name: task.name,
operationType: task.operationType,
cronExpression: task.cronExpression,
timezone: task.timezone,
targetSelector: task.targetSelector,
misfirePolicy: task.misfirePolicy,
overlapPolicy: task.overlapPolicy,
retryPolicy: task.retryPolicy,
enabled: task.enabled,
},
});
let id = 0;
const dispatch = vi.fn(async () => ({ outcome: 'succeeded' as const, jobIds: ['job-1'] }));
const coordinator = new SchedulerCoordinator({
db,
repository,
dispatch,
now: () => new Date('2026-07-30T01:00:00.000Z'),
id: () => `run-${++id}`,
});
const completed = await coordinator.tick();
expect(completed).toHaveLength(1);
expect(completed[0]?.id).toBe('queued-run');
expect(dispatch).toHaveBeenCalledTimes(1);
expect(repository.get('task-1')?.nextDueAt).toBe('2026-07-30T01:00:00.000Z');
db.close();
});
it('does not reschedule after stop while a tick is still running', async () => {
vi.useFakeTimers();
const { db, repository } = fixture();
let finishDispatch!: (value: { outcome: 'succeeded'; jobIds: string[] }) => void;
const dispatch = vi.fn(
() =>
new Promise<{ outcome: 'succeeded'; jobIds: string[] }>((resolve) => {
finishDispatch = resolve;
}),
);
const coordinator = new SchedulerCoordinator({
db,
repository,
dispatch,
now: () => new Date('2026-07-30T01:00:00.000Z'),
id: () => 'in-flight-run',
intervalMs: 1_000,
});
try {
coordinator.start();
expect(dispatch).toHaveBeenCalledTimes(1);
coordinator.stop();
finishDispatch({ outcome: 'succeeded', jobIds: ['job-1'] });
await vi.advanceTimersByTimeAsync(0);
expect(vi.getTimerCount()).toBe(0);
} finally {
coordinator.stop();
vi.useRealTimers();
db.close();
}
});
});
@@ -0,0 +1,326 @@
import { randomUUID } from 'node:crypto';
import type {
CreateScheduledTaskRequest,
ScheduledRun,
ScheduledRunOutcome,
ScheduledTask,
} from '@multi-simadmin/contracts';
import type { SqliteDatabase } from '../../infrastructure/database/database.js';
import { nextOccurrence, reconcileOccurrence } from './schedule-time.js';
import { ScheduledTaskRepository } from './scheduled-task-repository.js';
import { resolveOperationTargets, type ResolvedTarget } from './target-resolver.js';
export interface DispatchResult {
readonly outcome: ScheduledRunOutcome;
readonly reason?: string;
readonly jobIds: readonly string[];
}
export type ScheduledDispatch = (
task: ScheduledTask,
targets: readonly ResolvedTarget[],
run: ScheduledRun,
context: { readonly actor: string; readonly requestId: string },
) => Promise<DispatchResult>;
export interface SchedulerCoordinatorOptions {
readonly db: SqliteDatabase;
readonly repository: ScheduledTaskRepository;
readonly dispatch: ScheduledDispatch;
readonly now?: () => Date;
readonly id?: () => string;
readonly intervalMs?: number;
}
function snapshot(task: ScheduledTask): CreateScheduledTaskRequest {
return {
name: task.name,
operationType: task.operationType,
cronExpression: task.cronExpression,
timezone: task.timezone,
targetSelector: task.targetSelector,
...(task.effectiveStartAt ? { effectiveStartAt: task.effectiveStartAt } : {}),
...(task.effectiveEndAt ? { effectiveEndAt: task.effectiveEndAt } : {}),
misfirePolicy: task.misfirePolicy,
overlapPolicy: task.overlapPolicy,
retryPolicy: task.retryPolicy,
enabled: task.enabled,
};
}
export class SchedulerCoordinator {
private readonly now: () => Date;
private readonly id: () => string;
private readonly intervalMs: number;
private timer: ReturnType<typeof setTimeout> | undefined;
private started = false;
private lifecycle = 0;
constructor(private readonly options: SchedulerCoordinatorOptions) {
this.now = options.now ?? (() => new Date());
this.id = options.id ?? randomUUID;
this.intervalMs = options.intervalMs ?? 30_000;
}
start(): void {
if (this.started) return;
this.started = true;
const lifecycle = ++this.lifecycle;
const schedule = () => {
if (!this.started || lifecycle !== this.lifecycle) return;
this.timer = setTimeout(() => {
this.timer = undefined;
void this.tick().finally(schedule);
}, this.intervalMs);
this.timer.unref?.();
};
void this.tick().finally(schedule);
}
stop(): void {
this.started = false;
this.lifecycle += 1;
if (this.timer) clearTimeout(this.timer);
this.timer = undefined;
}
async runNow(taskId: string, actor: string, requestId: string): Promise<ScheduledRun> {
const task = this.options.repository.get(taskId);
if (!task) throw new Error('Scheduled task was not found');
return this.runOccurrence(task, this.now().toISOString(), 'manual', actor, requestId);
}
async tick(): Promise<readonly ScheduledRun[]> {
const now = this.now();
const nowIso = now.toISOString();
const completed: ScheduledRun[] = [];
for (const task of this.options.repository.list()) {
const queued = this.options.repository.getQueuedRun(task.id);
if (queued && !this.options.repository.hasActiveRun(task.id)) {
if (queued.scheduleVersion !== task.version) {
completed.push(
this.options.repository.finishRun(queued.id, {
outcome: 'needs-attention',
reason: 'Queued occurrence belongs to an outdated schedule version',
jobIds: [],
finishedAt: nowIso,
}),
);
} else if (!task.enabled) {
completed.push(
this.options.repository.finishRun(queued.id, {
outcome: 'skipped',
reason: 'Schedule was disabled while the occurrence was queued',
jobIds: [],
finishedAt: nowIso,
}),
);
} else {
const resolution = resolveOperationTargets(
this.options.db,
{ mode: 'fixed', instanceIds: queued.targetSnapshot },
task.operationType,
);
if (resolution.unavailableInstanceIds.length > 0) {
completed.push(
this.options.repository.finishRun(queued.id, {
outcome: 'needs-attention',
reason: this.capabilityReason(resolution.unavailableInstanceIds),
jobIds: [],
finishedAt: nowIso,
}),
);
continue;
}
completed.push(
await this.executeClaimed(
task,
queued,
resolution.targets,
'scheduled-automation',
`queue:${queued.id}`,
),
);
}
continue;
}
if (!task.enabled || !task.nextDueAt || task.nextDueAt > nowIso) continue;
const outsideWindow =
(task.effectiveStartAt !== undefined && nowIso < task.effectiveStartAt) ||
(task.effectiveEndAt !== undefined && nowIso > task.effectiveEndAt);
if (outsideWindow) {
const anchor =
task.effectiveStartAt && nowIso < task.effectiveStartAt
? new Date(task.effectiveStartAt)
: now;
const nextDueAt = nextOccurrence(task.cronExpression, anchor);
const claimed = this.options.repository.claimScheduledOccurrence({
id: this.id(),
scheduledTaskId: task.id,
scheduleVersion: task.version,
dueAt: task.nextDueAt,
nextDueAt,
claimedAt: nowIso,
taskSnapshot: snapshot(task),
overlapPolicy: task.overlapPolicy,
resolveTargets: () => ({ targets: [] }),
terminal: {
outcome: 'skipped',
reason: 'Occurrence is outside the effective window',
},
});
if (claimed) completed.push(claimed.run);
continue;
}
const exact = task.nextDueAt === nowIso;
const reconciled = exact
? {
action: 'run' as const,
dueAt: task.nextDueAt,
nextDueAt: nextOccurrence(task.cronExpression, now),
}
: reconcileOccurrence(
{
cronExpression: task.cronExpression,
nextDueAt: task.nextDueAt,
misfirePolicy: task.misfirePolicy,
},
now,
);
if (reconciled.action === 'wait') continue;
if (reconciled.action === 'skip') {
const claimed = this.options.repository.claimScheduledOccurrence({
id: this.id(),
scheduledTaskId: task.id,
scheduleVersion: task.version,
dueAt: reconciled.dueAt,
nextDueAt: reconciled.nextDueAt,
claimedAt: nowIso,
taskSnapshot: snapshot(task),
overlapPolicy: task.overlapPolicy,
resolveTargets: () => ({ targets: [] }),
terminal: {
outcome: 'skipped',
reason: 'Missed occurrence skipped by task policy',
},
});
if (claimed) completed.push(claimed.run);
continue;
}
const claimed = this.options.repository.claimScheduledOccurrence({
id: this.id(),
scheduledTaskId: task.id,
scheduleVersion: task.version,
dueAt: reconciled.dueAt,
nextDueAt: reconciled.nextDueAt,
claimedAt: nowIso,
taskSnapshot: snapshot(task),
overlapPolicy: task.overlapPolicy,
resolveTargets: () => {
const resolution = resolveOperationTargets(
this.options.db,
task.targetSelector,
task.operationType,
);
return {
targets: resolution.targets,
targetSnapshot: resolution.targetSnapshot,
...(resolution.unavailableInstanceIds.length > 0
? {
attentionReason: this.capabilityReason(resolution.unavailableInstanceIds),
}
: {}),
};
},
});
if (!claimed || claimed.disposition === 'queued') continue;
if (claimed.disposition === 'finished') {
completed.push(claimed.run);
continue;
}
completed.push(
await this.executeClaimed(
task,
claimed.run,
claimed.targets,
'scheduled-automation',
`schedule:${task.id}:${reconciled.dueAt}`,
),
);
}
return completed;
}
private async runOccurrence(
task: ScheduledTask,
dueAt: string,
triggerSource: 'scheduled' | 'manual',
actor: string,
requestId: string,
): Promise<ScheduledRun> {
const resolution = resolveOperationTargets(
this.options.db,
task.targetSelector,
task.operationType,
);
const claimed = this.options.repository.claimOccurrence({
id: this.id(),
scheduledTaskId: task.id,
scheduleVersion: task.version,
dueAt,
claimedAt: this.now().toISOString(),
triggerSource,
targetSnapshot: resolution.targetSnapshot,
taskSnapshot: snapshot(task),
});
if (!claimed) throw new Error('Scheduled occurrence was already claimed');
if (resolution.unavailableInstanceIds.length > 0)
return this.options.repository.finishRun(claimed.id, {
outcome: 'needs-attention',
reason: this.capabilityReason(resolution.unavailableInstanceIds),
jobIds: [],
finishedAt: this.now().toISOString(),
});
return this.executeClaimed(task, claimed, resolution.targets, actor, requestId);
}
private capabilityReason(instanceIds: readonly string[]): string {
return `Required capability is unavailable for: ${instanceIds.join(', ')}`;
}
private async executeClaimed(
task: ScheduledTask,
claimed: ScheduledRun,
targets: readonly ResolvedTarget[],
actor: string,
requestId: string,
): Promise<ScheduledRun> {
const now = this.now().toISOString();
if (targets.length === 0)
return this.options.repository.finishRun(claimed.id, {
outcome: 'no-targets',
reason: 'No enabled instances matched the current selector',
jobIds: [],
finishedAt: now,
});
this.options.repository.startRun(claimed.id, now);
try {
const result = await this.options.dispatch(task, targets, claimed, { actor, requestId });
return this.options.repository.finishRun(claimed.id, {
outcome: result.outcome,
...(result.reason ? { reason: result.reason } : {}),
jobIds: result.jobIds,
finishedAt: this.now().toISOString(),
});
} catch {
return this.options.repository.finishRun(claimed.id, {
outcome: 'failed',
reason: 'Scheduled dispatch failed',
jobIds: [],
finishedAt: this.now().toISOString(),
});
}
}
}
@@ -0,0 +1,85 @@
import Database from 'better-sqlite3';
import { describe, expect, it } from 'vitest';
import { migrateDatabase } from '../../infrastructure/database/migrations.js';
import { resolveOperationTargets, resolveTargets } from './target-resolver.js';
function fixture() {
const db = new Database(':memory:');
db.pragma('foreign_keys = ON');
migrateDatabase(db);
const now = '2026-07-30T00:00:00.000Z';
const insert = db.prepare(
'INSERT INTO instances (id,name,base_url,enabled,config_revision,created_at,updated_at) VALUES (?,?,?,?,?,?,?)',
);
insert.run('a', 'Alpha', 'http://10.0.0.1', 1, 3, now, now);
insert.run('b', 'Beta', 'http://10.0.0.2', 1, 2, now, now);
insert.run('c', 'Disabled', 'http://10.0.0.3', 0, 1, now, now);
const tag = db.prepare('INSERT INTO instance_tags (instance_id,tag,created_at) VALUES (?,?,?)');
tag.run('a', 'lab', now);
tag.run('a', 'east', now);
tag.run('b', 'lab', now);
tag.run('c', 'east', now);
return db;
}
describe('resolveTargets', () => {
it('resolves enabled fixed targets in selector order with current revisions', () => {
const db = fixture();
expect(resolveTargets(db, { mode: 'fixed', instanceIds: ['b', 'c', 'a'] })).toEqual([
{ id: 'b', revision: 2 },
{ id: 'a', revision: 3 },
]);
db.close();
});
it('resolves dynamic all and any tag matches on every call', () => {
const db = fixture();
expect(resolveTargets(db, { mode: 'tags', match: 'all', tags: ['lab', 'east'] })).toEqual([
{ id: 'a', revision: 3 },
]);
expect(resolveTargets(db, { mode: 'tags', match: 'any', tags: ['lab', 'east'] })).toEqual([
{ id: 'a', revision: 3 },
{ id: 'b', revision: 2 },
]);
db.prepare(
'INSERT INTO instances (id,name,base_url,enabled,config_revision,created_at,updated_at) VALUES (?,?,?,?,?,?,?)',
).run(
'd',
'Dynamic',
'http://10.0.0.4',
1,
1,
'2026-07-30T00:00:00.000Z',
'2026-07-30T00:00:00.000Z',
);
db.prepare('INSERT INTO instance_tags (instance_id,tag,created_at) VALUES (?,?,?)').run(
'd',
'east',
'2026-07-30T00:00:00.000Z',
);
expect(
resolveTargets(db, { mode: 'tags', match: 'any', tags: ['east'] }).map((x) => x.id),
).toEqual(['a', 'd']);
db.close();
});
it('separates targets whose required operation capability is unavailable', () => {
const db = fixture();
const now = '2026-07-30T00:00:00.000Z';
db.prepare(
'INSERT INTO capabilities (instance_id,operation_id,state,observed_at,created_at,updated_at) VALUES (?,?,?,?,?,?)',
).run('a', 'postServiceRestart', 'supported', now, now, now);
db.prepare(
'INSERT INTO capabilities (instance_id,operation_id,state,observed_at,created_at,updated_at) VALUES (?,?,?,?,?,?)',
).run('b', 'postServiceRestart', 'unsupported', now, now, now);
expect(
resolveOperationTargets(db, { mode: 'fixed', instanceIds: ['a', 'b'] }, 'restart-service'),
).toEqual({
targets: [{ id: 'a', revision: 3 }],
targetSnapshot: ['a', 'b'],
unavailableInstanceIds: ['b'],
});
db.close();
});
});
@@ -0,0 +1,91 @@
import type { ScheduledOperationType, ScheduleTargetSelector } from '@multi-simadmin/contracts';
import type { SqliteDatabase } from '../../infrastructure/database/database.js';
export interface ResolvedTarget {
readonly id: string;
readonly revision: number;
}
interface TargetRow {
id: string;
config_revision: number;
}
export interface OperationTargetResolution {
readonly targets: readonly ResolvedTarget[];
readonly targetSnapshot: readonly string[];
readonly unavailableInstanceIds: readonly string[];
}
const operationIds: Record<ScheduledOperationType, string> = {
'restart-service': 'postServiceRestart',
'reboot-system': 'postSystemReboot',
'send-sms': 'postSmsSend',
};
export function resolveTargets(
db: SqliteDatabase,
selector: ScheduleTargetSelector,
): readonly ResolvedTarget[] {
if (selector.mode === 'fixed') {
const placeholders = selector.instanceIds.map(() => '?').join(',');
const rows = db
.prepare(
`SELECT id, config_revision FROM instances
WHERE enabled = 1 AND id IN (${placeholders})`,
)
.all(...selector.instanceIds) as TargetRow[];
const byId = new Map(rows.map((row) => [row.id, row]));
return selector.instanceIds.flatMap((id) => {
const row = byId.get(id);
return row ? [{ id: row.id, revision: row.config_revision }] : [];
});
}
const placeholders = selector.tags.map(() => '?').join(',');
const comparison = selector.match === 'all' ? '= ?' : '> 0';
const parameters: Array<string | number> = [...selector.tags];
if (selector.match === 'all') parameters.push(selector.tags.length);
const rows = db
.prepare(
`SELECT i.id, i.config_revision
FROM instances i
JOIN instance_tags t ON t.instance_id = i.id
WHERE i.enabled = 1 AND t.tag IN (${placeholders})
GROUP BY i.id, i.config_revision
HAVING COUNT(DISTINCT t.tag) ${comparison}
ORDER BY i.id ASC`,
)
.all(...parameters) as TargetRow[];
return rows.map((row) => ({ id: row.id, revision: row.config_revision }));
}
export function resolveOperationTargets(
db: SqliteDatabase,
selector: ScheduleTargetSelector,
operationType: ScheduledOperationType,
): OperationTargetResolution {
const candidates = resolveTargets(db, selector);
if (candidates.length === 0)
return { targets: [], targetSnapshot: [], unavailableInstanceIds: [] };
const placeholders = candidates.map(() => '?').join(',');
const rows = db
.prepare(
`SELECT instance_id, state FROM capabilities
WHERE operation_id = ? AND instance_id IN (${placeholders})`,
)
.all(operationIds[operationType], ...candidates.map((target) => target.id)) as Array<{
instance_id: string;
state: string;
}>;
const states = new Map(rows.map((row) => [row.instance_id, row.state]));
const availableStates = new Set(['supported', 'auth-required', 'degraded']);
return {
targets: candidates.filter((target) => availableStates.has(states.get(target.id) ?? 'unknown')),
targetSnapshot: candidates.map((target) => target.id),
unavailableInstanceIds: candidates
.filter((target) => !availableStates.has(states.get(target.id) ?? 'unknown'))
.map((target) => target.id),
};
}
@@ -4,12 +4,18 @@ import {
InstanceMessageService,
MessageServiceError,
parseMessageList,
validMessageContent,
} from './instance-message-service.js';
const instance = { id: 'alpha', origin: 'http://192.168.1.10:8080' };
const instances = { get: vi.fn(async (id: string) => (id === 'alpha' ? instance : undefined)) };
describe('InstanceMessageService', () => {
it('accepts the automation contract maximum of 2,000 SMS characters', () => {
expect(validMessageContent('字'.repeat(2_000))).toBe(true);
expect(validMessageContent('字'.repeat(2_001))).toBe(false);
});
it('parses a bounded explicit message allowlist and never exposes pdu or excess fields', () => {
const messages = parseMessageList(
{
@@ -51,6 +57,43 @@ describe('InstanceMessageService', () => {
expect(JSON.stringify(messages)).not.toContain('secret');
});
it('accepts the current SimAdmin SMS shape without the removed transport field', () => {
expect(
parseMessageList(
{
status: 200,
headers: {},
body: JSON.stringify({
status: 'success',
data: {
messages: [
{
id: 42,
direction: 'incoming',
phone_number: '10086',
content: 'current payload',
timestamp: '2026-07-29 12:30:00',
status: 'received',
},
],
},
}),
},
10,
),
).toEqual([
{
id: '42',
direction: 'incoming',
phoneNumber: '10086',
content: 'current payload',
timestamp: '2026-07-29 12:30:00',
status: 'received',
transport: 'modem',
},
]);
});
it('uses exact owner, bounded query and matching optional session cookie', async () => {
const sessions = new InstanceSessionStore();
sessions.set('alpha', instance.origin, 'simadmin_session=opaque');
@@ -55,8 +55,8 @@ export function validPhoneNumber(value: string): boolean {
export function validMessageContent(value: string): boolean {
return (
value.length >= 1 &&
value.length <= 1600 &&
Buffer.byteLength(value, 'utf8') <= 6400 &&
value.length <= 2000 &&
Buffer.byteLength(value, 'utf8') <= 8000 &&
!/[\u0000-\u0008\u000b\u000c\u000e-\u001f\u007f]/u.test(value)
);
}
@@ -87,10 +87,12 @@ export function parseMessageList(
const value = record(item);
const id = bounded(value?.id, 128);
const phoneNumber = bounded(value?.phone_number, 32);
const content = bounded(value?.content, 1600);
const content = bounded(value?.content, 2000);
const timestamp = bounded(value?.timestamp, 64);
const status = bounded(value?.status, 32);
const transport = bounded(value?.transport, 32);
// Current SimAdmin SmsMessage has no transport field; older captures sometimes did.
// Keep the aggregate contract stable without dropping every current message.
const transport = bounded(value?.transport, 32) ?? 'modem';
const rawDirection = bounded(value?.direction, 16);
if (
!id ||
@@ -98,8 +100,7 @@ export function parseMessageList(
!validPhoneNumber(phoneNumber) ||
content === undefined ||
timestamp === undefined ||
status === undefined ||
transport === undefined
status === undefined
)
continue;
const direction: MessageDirection =
@@ -133,6 +134,11 @@ export class InstanceMessageService {
readonly instances: InstanceService;
readonly sessions: InstanceSessionStore;
readonly request: UpstreamSessionClientOptions['request'];
readonly ensureSession?: (
instanceId: string,
origin: string,
force?: boolean,
) => Promise<void>;
},
) {}
@@ -142,7 +148,14 @@ export class InstanceMessageService {
const session = this.options.sessions.sessionFor(instanceId);
if (session && session.origin !== instance.origin)
throw new MessageServiceError('SESSION_INVALID');
return { instance, session };
if (!session && this.options.ensureSession) {
try {
await this.options.ensureSession(instanceId, instance.origin);
} catch {
// No saved credential means this may be a passwordless instance; read anonymously.
}
}
return { instance, session: this.options.sessions.sessionFor(instanceId) };
}
async list(
@@ -163,11 +176,17 @@ export class InstanceMessageService {
throw new MessageServiceError('VALIDATION_FAILED');
const { instance, session } = await this.owner(instanceId);
const direction = query.direction ? `&direction=${query.direction}` : '';
const response = await this.options.request({
url: `${instance.origin}/api/sms/list?limit=${query.limit}&offset=${query.offset}${direction}`,
method: 'GET',
headers: { accept: 'application/json', ...(session ? { cookie: session.cookie } : {}) },
});
const request = (cookie?: string) =>
this.options.request({
url: `${instance.origin}/api/sms/list?limit=${query.limit}&offset=${query.offset}${direction}`,
method: 'GET',
headers: { accept: 'application/json', ...(cookie ? { cookie } : {}) },
});
let response = await request(session?.cookie);
if ([401, 403].includes(response.status) && this.options.ensureSession) {
await this.options.ensureSession(instanceId, instance.origin, true);
response = await request(this.options.sessions.sessionFor(instanceId)?.cookie);
}
return { messages: parseMessageList(response, query.limit) };
}
@@ -175,16 +194,22 @@ export class InstanceMessageService {
if (!validPhoneNumber(input.phoneNumber) || !validMessageContent(input.content))
throw new MessageServiceError('VALIDATION_FAILED');
const { instance, session } = await this.owner(instanceId);
const response = await this.options.request({
url: `${instance.origin}/api/sms/send`,
method: 'POST',
headers: {
accept: 'application/json',
'content-type': 'application/json',
...(session ? { cookie: session.cookie } : {}),
},
sms: { phoneNumber: input.phoneNumber, content: input.content },
});
const request = (cookie?: string) =>
this.options.request({
url: `${instance.origin}/api/sms/send`,
method: 'POST',
headers: {
accept: 'application/json',
'content-type': 'application/json',
...(cookie ? { cookie } : {}),
},
sms: { phoneNumber: input.phoneNumber, content: input.content },
});
let response = await request(session?.cookie);
if ([401, 403].includes(response.status) && this.options.ensureSession) {
await this.options.ensureSession(instanceId, instance.origin, true);
response = await request(this.options.sessions.sessionFor(instanceId)?.cookie);
}
parseSendSuccess(response);
return { sent: true };
}
@@ -130,6 +130,7 @@ describe('SecureOperationExecution generic R2 slice', () => {
query: '',
contentType: '',
body: undefined,
instanceId: 'i-1',
});
await expectCode(
execution.execute(
@@ -244,10 +245,11 @@ describe('SecureOperationExecution R3 restart slice', () => {
query: '',
contentType: '',
body: undefined,
instanceId: 'i-1',
});
expect(db.prepare('SELECT risk_level FROM jobs WHERE id=?').get(job.id)).toEqual({
risk_level: 'R3',
});
expect(
db.prepare('SELECT risk_level FROM jobs WHERE id=?').get(job.id),
).toEqual({ risk_level: 'R3' });
});
it('prepares and executes system reboot only with fixed delay_seconds=3', async () => {
@@ -283,6 +285,7 @@ describe('SecureOperationExecution R3 restart slice', () => {
query: '',
contentType: 'application/json',
body: '{"delay_seconds":3}',
instanceId: 'i-1',
});
});
});
@@ -71,6 +71,8 @@ export interface SafeOperationTransportRequest {
readonly contentType: string;
/** Serialized JSON body for audited JSON operations; undefined for zero-body POSTs. */
readonly body: string | undefined;
/** Bound target used only for origin-scoped session attachment; never logged. */
readonly instanceId: string;
}
export interface SafeOperationTransport {
request(request: SafeOperationTransportRequest): Promise<{ readonly status: number }>;
@@ -199,7 +201,8 @@ export class SecureOperationExecution {
let contentType = '';
let serializedBody: string | undefined;
if (ZERO_BODY_OPERATIONS.has(descriptor.operationId)) {
if (descriptor.requestContentType !== 'none' || parameters.fields.length !== 0) this.validation();
if (descriptor.requestContentType !== 'none' || parameters.fields.length !== 0)
this.validation();
} else if (descriptor.operationId === SYSTEM_REBOOT_OPERATION) {
if (descriptor.requestContentType !== 'application/json' || parameters.fields.length !== 1)
this.validation();
@@ -412,6 +415,7 @@ export class SecureOperationExecution {
bound.operation_id === SYSTEM_REBOOT_OPERATION
? JSON.stringify({ delay_seconds: SYSTEM_REBOOT_DELAY_SECONDS })
: undefined,
instanceId: bound.target_instance_id,
});
state = response.status >= 200 && response.status < 300 ? 'succeeded' : 'failed';
code = state === 'succeeded' ? 'UPSTREAM_SUCCEEDED' : 'UPSTREAM_REJECTED';
@@ -1,6 +1,11 @@
import { describe, expect, it } from 'vitest';
import { InstanceResourceService, parseSim, parseStats } from './instance-resource-service.js';
import {
InstanceResourceService,
parseHealth,
parseSim,
parseStats,
} from './instance-resource-service.js';
const response = (body: unknown) => ({
status: 200,
@@ -27,6 +32,27 @@ describe('instance resource allowlist parsing', () => {
).toEqual({ cpuPercent: 23.4, memoryPercent: 67.8, maxTemperatureCelsius: 52.6 });
});
it('falls back to version metadata exposed by stats on older SimAdmin builds', () => {
expect(
parseStats(
response({
data: {
cpu_load: { load_percent: 10 },
system: { app_version: '1.8.7', architecture: 'aarch64' },
},
}),
),
).toMatchObject({ cpuPercent: 10, version: '1.8.7', platform: 'aarch64' });
});
it('extracts the upstream SimAdmin version from the health endpoint', () => {
expect(
parseHealth(
response({ status: 'ok', version: '1.9.4', platform: 'linux-aarch64', secret: 'drop' }),
),
).toEqual({ version: '1.9.4', platform: 'linux-aarch64' });
});
it('extracts, validates, deduplicates and bounds phone numbers only', () => {
expect(
parseSim(
@@ -51,7 +77,9 @@ describe('instance resource allowlist parsing', () => {
return response(
request.url.endsWith('/api/stats')
? { data: { cpu_load: { load_percent: 12 }, memory: { used_percent: 34 } } }
: { data: { phone_numbers: ['13800000000'] } },
: request.url.endsWith('/api/sim')
? { data: { phone_numbers: ['13800000000'] } }
: { status: 'ok', version: '2.0.1', platform: 'linux' },
);
},
});
@@ -59,10 +87,13 @@ describe('instance resource allowlist parsing', () => {
cpuPercent: 12,
memoryPercent: 34,
phoneNumbers: ['13800000000'],
version: '2.0.1',
platform: 'linux',
});
expect(requests.map((request) => request.headers)).toEqual([
{ accept: 'application/json' },
{ accept: 'application/json' },
{ accept: 'application/json' },
]);
});
@@ -10,6 +10,8 @@ export interface InstanceResources {
readonly memoryPercent?: number;
readonly maxTemperatureCelsius?: number;
readonly phoneNumbers?: readonly string[];
readonly version?: string;
readonly platform?: string;
}
const MAX_BODY_BYTES = 32_768;
@@ -32,6 +34,31 @@ function data(response: UpstreamResponse): Record<string, unknown> | undefined {
}
}
const safeText = (value: unknown, maximum = 128): string | undefined =>
typeof value === 'string' &&
value.length > 0 &&
value.length <= maximum &&
!/[\u0000-\u001f\u007f]/u.test(value)
? value
: undefined;
export function parseHealth(response: UpstreamResponse): InstanceResources {
if (response.status < 200 || response.status >= 300) return {};
if (Buffer.byteLength(response.body, 'utf8') > MAX_BODY_BYTES) return {};
try {
const root = record(JSON.parse(response.body));
if (!root) return {};
const version = safeText(root.version);
const platform = safeText(root.platform);
return {
...(version ? { version } : {}),
...(platform ? { platform } : {}),
};
} catch {
return {};
}
}
export function parseStats(response: UpstreamResponse): InstanceResources {
const value = data(response);
if (!value) return {};
@@ -43,10 +70,20 @@ export function parseStats(response: UpstreamResponse): InstanceResources {
.filter((item): item is number => item !== undefined)
: [];
const maxTemperatureCelsius = temperatures.length > 0 ? Math.max(...temperatures) : undefined;
const system = record(value.system);
const version =
safeText(value.version) ??
safeText(value.current_version) ??
safeText(system?.version) ??
safeText(system?.app_version);
const platform =
safeText(value.platform) ?? safeText(system?.platform) ?? safeText(system?.architecture);
return {
...(cpuPercent === undefined ? {} : { cpuPercent }),
...(memoryPercent === undefined ? {} : { memoryPercent }),
...(maxTemperatureCelsius === undefined ? {} : { maxTemperatureCelsius }),
...(version ? { version } : {}),
...(platform ? { platform } : {}),
};
}
@@ -69,6 +106,11 @@ export class InstanceResourceService {
readonly instances: InstanceService;
readonly sessions: InstanceSessionStore;
readonly request: UpstreamSessionClientOptions['request'];
readonly ensureSession?: (
instanceId: string,
origin: string,
force?: boolean,
) => Promise<void>;
},
) {}
@@ -79,19 +121,48 @@ export class InstanceResourceService {
]);
if (!instance) return {};
if (session && session.origin !== instance.origin) return {};
const get = (path: '/api/stats' | '/api/sim') =>
this.options.request({
url: `${instance.origin}${path}`,
method: 'GET',
headers: {
accept: 'application/json',
...(session ? { cookie: session.cookie } : {}),
},
});
const [stats, sim] = await Promise.allSettled([get('/api/stats'), get('/api/sim')]);
if (!session && this.options.ensureSession) {
try {
await this.options.ensureSession(instanceId, instance.origin);
} catch {
// Passwordless and temporarily unavailable credentials still permit anonymous reads.
}
}
const paths = ['/api/stats', '/api/sim', '/api/health'] as const;
const readAll = () => {
const activeSession = this.options.sessions.sessionFor(instanceId);
return Promise.allSettled(
paths.map((path) =>
this.options.request({
url: `${instance.origin}${path}`,
method: 'GET',
headers: {
accept: 'application/json',
...(activeSession?.origin === instance.origin
? { cookie: activeSession.cookie }
: {}),
},
}),
),
);
};
let results = await readAll();
const unauthorized = results.some(
(result) => result.status === 'fulfilled' && [401, 403].includes(result.value.status),
);
if (unauthorized && this.options.ensureSession) {
try {
await this.options.ensureSession(instanceId, instance.origin, true);
results = await readAll();
} catch {
// Return the safe partial result when re-authentication is unavailable.
}
}
const [stats, sim, health] = results;
return {
...(stats.status === 'fulfilled' ? parseStats(stats.value) : {}),
...(sim.status === 'fulfilled' ? parseSim(sim.value) : {}),
...(stats?.status === 'fulfilled' ? parseStats(stats.value) : {}),
...(sim?.status === 'fulfilled' ? parseSim(sim.value) : {}),
...(health?.status === 'fulfilled' ? parseHealth(health.value) : {}),
};
}
}
+242 -11
View File
@@ -1,6 +1,7 @@
import Database from 'better-sqlite3';
import { afterEach, describe, expect, it } from 'vitest';
import { buildControlPlaneApp } from './control-plane.js';
import { ScheduledTaskRepository } from './application/automation/scheduled-task-repository.js';
import { migrateDatabase } from './infrastructure/database/migrations.js';
import { UpstreamError } from './infrastructure/transport/upstream-error.js';
import { SafeInstanceTransport } from './infrastructure/transport/safe-instance-transport.js';
@@ -12,17 +13,83 @@ afterEach(async () => {
for (const db of dbs.splice(0)) db.close();
});
class Store implements SecretStore {
async set() {
return '';
private readonly values = new Map<string, string>();
async set(key: { instanceId: string; purpose: string; slot?: string }, value: string) {
const account = Buffer.from(
JSON.stringify(
key.slot === undefined
? [key.instanceId, key.purpose]
: [key.instanceId, key.purpose, key.slot],
),
'utf8',
).toString('base64url');
const reference = `keychain://multi-simadmin/${account}`;
this.values.set(reference, value);
return reference;
}
async get() {
return '[REDACTED]';
async get(reference: string) {
return this.values.get(reference);
}
async delete() {
return false;
async delete(reference: string) {
return this.values.delete(reference);
}
}
describe('buildControlPlaneApp', () => {
it('reconciles interrupted scheduled runs before starting the scheduler', async () => {
const db = new Database(':memory:');
db.pragma('foreign_keys=ON');
migrateDatabase(db);
dbs.push(db);
const repository = new ScheduledTaskRepository(db);
const task = {
name: 'Interrupted restart',
operationType: 'restart-service' as const,
cronExpression: '0 9 * * *',
timezone: 'Asia/Shanghai' as const,
targetSelector: { mode: 'fixed' as const, instanceIds: ['instance-a'] },
misfirePolicy: 'skip' as const,
overlapPolicy: 'skip' as const,
retryPolicy: { maxRetries: 0, intervalSeconds: 60 },
enabled: false,
};
repository.create({
id: 'task-1',
task,
createdBy: 'operator',
now: '2026-07-30T00:00:00.000Z',
});
repository.claimOccurrence({
id: 'run-1',
scheduledTaskId: 'task-1',
scheduleVersion: 1,
dueAt: '2026-07-30T01:00:00.000Z',
claimedAt: '2026-07-30T01:00:00.000Z',
triggerSource: 'scheduled',
targetSnapshot: [],
taskSnapshot: task,
});
repository.startRun('run-1', '2026-07-30T01:00:01.000Z');
const app = buildControlPlaneApp({
db,
store: new Store(),
upstream: {
get: async () => ({ status: 200, headers: {}, body: '' }),
request: async () => ({ status: 200, headers: {}, body: '' }),
postNetworkRegisterAuto: async () => ({ status: 200 }),
postServiceRestart: async () => ({ status: 200 }),
postSystemReboot: async () => ({ status: 200 }),
},
now: () => new Date('2026-07-30T02:00:00.000Z'),
});
expect(repository.getRun('run-1')).toMatchObject({
outcome: 'needs-attention',
finishedAt: '2026-07-30T02:00:00.000Z',
});
await app.close();
});
it('registers the durable event route fail-closed when no authentication dependency is supplied', async () => {
const db = new Database(':memory:');
db.pragma('foreign_keys=ON');
@@ -127,14 +194,178 @@ describe('buildControlPlaneApp', () => {
url: '/api/v1/instances/' + created.json().id + '/login',
payload: { password: 'never-leak-this' },
});
expect(insecureLogin.statusCode).toBe(400);
// Literal private HTTP origins are allowed for LAN instances; secrets still never leak.
expect(insecureLogin.statusCode).toBe(200);
expect(insecureLogin.json()).toMatchObject({
title: 'Bad Request',
status: 400,
code: 'UPSTREAM_INSECURE_AUTH',
authenticated: true,
instanceId: created.json().id,
});
expect(insecureLogin.body).not.toContain('never-leak-this');
expect(insecureLogin.body).not.toContain('192.168.1.10');
expect(insecureLogin.body).not.toContain('simadmin_session');
await app.close();
});
it('automatically logs in with saved credentials before aggregate messages and resource reads', async () => {
const db = new Database(':memory:');
db.pragma('foreign_keys=ON');
migrateDatabase(db);
dbs.push(db);
const calls: Array<{ url: string; cookie?: string }> = [];
const app = buildControlPlaneApp({
db,
store: new Store(),
upstream: {
get: async () => ({ status: 200, headers: {}, body: '' }),
request: async (request) => {
calls.push({
url: request.url,
...(request.headers.cookie ? { cookie: request.headers.cookie } : {}),
});
if (request.url.endsWith('/api/auth/login'))
return {
status: 200,
headers: { 'set-cookie': 'simadmin_session=aggregate-session' },
body: '{"status":"success"}',
};
if (request.url.includes('/api/sms/list'))
return {
status: 200,
headers: {},
body: JSON.stringify({
status: 'success',
data: {
messages: [
{
id: 7,
direction: 'incoming',
phone_number: '10086',
content: 'hi',
timestamp: '2026-07-29 12:00:00',
status: 'received',
},
],
},
}),
};
if (request.url.endsWith('/api/health'))
return {
status: 200,
headers: {},
body: '{"status":"ok","version":"2.3.4","platform":"linux"}',
};
return { status: 200, headers: {}, body: '{"status":"success","data":{}}' };
},
postNetworkRegisterAuto: async () => ({ status: 200 }),
postServiceRestart: async () => ({ status: 200 }),
postSystemReboot: async () => ({ status: 200 }),
},
});
const created = await app.inject({
method: 'POST',
url: '/api/v1/instances',
payload: {
name: 'Protected',
origin: 'http://192.168.1.20:3000',
password: { action: 'set', password: '[REDACTED]' },
},
});
const id = created.json().id as string;
const messages = await app.inject({
method: 'GET',
url: `/api/v1/instances/${id}/messages?limit=1&offset=0`,
});
expect(messages.statusCode).toBe(200);
expect(messages.json().messages).toEqual([
expect.objectContaining({ id: '7', phoneNumber: '10086', content: 'hi' }),
]);
const resources = await app.inject({ method: 'GET', url: `/api/v1/instances/${id}/resources` });
expect(resources.statusCode).toBe(200);
expect(resources.json()).toMatchObject({ version: '2.3.4', platform: 'linux' });
expect(calls.filter((call) => call.url.endsWith('/api/auth/login'))).toHaveLength(1);
for (const call of calls.filter((item) => !item.url.endsWith('/api/auth/login')))
expect(call.cookie).toBe('simadmin_session=aggregate-session');
await app.close();
});
it('attaches an origin-bound session cookie when executing restart operations', async () => {
const db = new Database(':memory:');
db.pragma('foreign_keys=ON');
migrateDatabase(db);
dbs.push(db);
const calls: unknown[] = [];
const app = buildControlPlaneApp({
db,
store: new Store(),
upstream: {
get: async () => ({ status: 200, headers: {}, body: '' }),
request: async (request) => {
calls.push({ kind: 'auth', request });
if (request.url.endsWith('/api/auth/login'))
return {
status: 200,
headers: { 'set-cookie': 'simadmin_session=opaque-token' },
body: '',
};
return { status: 200, headers: {}, body: '' };
},
postNetworkRegisterAuto: async () => ({ status: 200 }),
postServiceRestart: async (origin, cookie) => {
calls.push({ kind: 'restart', origin, cookie });
return { status: 200 };
},
postSystemReboot: async (origin, delaySeconds, cookie) => {
calls.push({ kind: 'reboot', origin, delaySeconds, cookie });
return { status: 200 };
},
},
now: () => new Date('2026-07-16T12:00:00.000Z'),
});
const created = await app.inject({
method: 'POST',
url: '/api/v1/instances',
payload: {
name: 'Restart Target',
origin: 'http://192.168.1.10:8080',
password: { action: 'set', password: '[REDACTED]' },
},
});
expect(created.statusCode).toBe(201);
const instanceId = created.json().id as string;
const revision = created.json().revision as number;
const prepared = await app.inject({
method: 'POST',
url: '/api/v1/operations/prepare',
payload: {
operationId: 'postServiceRestart',
targets: [{ instanceId, revision }],
parameters: {
parameterSchemaId: 'simadmin.58e2204.postServiceRestart.parameters.v1',
fields: [],
},
},
});
expect(prepared.statusCode).toBe(200);
const executed = await app.inject({
method: 'POST',
url: '/api/v1/operations/execute',
payload: {
preparationId: prepared.json().id,
confirmationToken: prepared.json().confirmationToken,
},
});
expect(executed.statusCode).toBe(202);
expect(executed.json()).toMatchObject({
operationId: 'postServiceRestart',
status: 'succeeded',
});
expect(calls).toContainEqual({
kind: 'restart',
origin: 'http://192.168.1.10:8080',
cookie: 'simadmin_session=opaque-token',
});
// Auth request body uses the redacted marker; the actual secret must never appear.
expect(JSON.stringify(calls)).toContain('"[REDACTED]"');
expect(JSON.stringify(calls)).not.toMatch(/never-leak|password":"[^[]/);
await app.close();
});
+99 -11
View File
@@ -32,12 +32,21 @@ import { InstanceResourceService } from './application/resources/instance-resour
import { ConsoleAuthService } from './application/auth/console-auth-service.js';
import { registerConsoleAuth } from './interface/http/console-auth-routes.js';
import { InstanceMessageService } from './application/messages/instance-message-service.js';
import { ScheduledTaskRepository } from './application/automation/scheduled-task-repository.js';
import { ScheduledTaskService } from './application/automation/scheduled-task-service.js';
import { registerAutomationRoutes } from './interface/http/automation-routes.js';
import { ScheduledOperationDispatcher } from './application/automation/scheduled-operation-dispatcher.js';
import { SchedulerCoordinator } from './application/automation/scheduler-coordinator.js';
export interface SafeControlPlaneUpstream extends ConnectionTransport {
request: UpstreamSessionClientOptions['request'];
postNetworkRegisterAuto(origin: string): Promise<{ readonly status: number }>;
postServiceRestart(origin: string): Promise<{ readonly status: number }>;
postSystemReboot(origin: string, delaySeconds: number): Promise<{ readonly status: number }>;
postNetworkRegisterAuto(origin: string, cookie?: string): Promise<{ readonly status: number }>;
postServiceRestart(origin: string, cookie?: string): Promise<{ readonly status: number }>;
postSystemReboot(
origin: string,
delaySeconds: number,
cookie?: string,
): Promise<{ readonly status: number }>;
}
export interface ControlPlaneOptions {
readonly db: Database.Database;
@@ -54,6 +63,13 @@ export function buildControlPlaneApp(options: ControlPlaneOptions): ControlPlane
const eventJournal = new EventJournal(options.db);
const jobs = new JobQueryService(options.db);
const audit = new AuditQueryService(options.db);
const scheduledTasks = new ScheduledTaskRepository(options.db);
scheduledTasks.reconcileInterruptedRuns((options.now?.() ?? new Date()).toISOString());
const automation = new ScheduledTaskService({
repository: scheduledTasks,
store: options.store,
...(options.now ? { now: options.now } : {}),
});
const instances = options.now
? new InstanceService({ db: options.db, store: options.store, now: options.now })
: new InstanceService({ db: options.db, store: options.store });
@@ -67,41 +83,91 @@ export function buildControlPlaneApp(options: ControlPlaneOptions): ControlPlane
: new ConnectionProbe({ db: options.db, instances, transport: options.upstream });
const sessions = new InstanceSessionStore();
const client = new UpstreamSessionClient({ sessions, request: options.upstream.request });
const resolver = new InstanceCredentialResolver({ db: options.db, store: options.store });
const login = options.now
? new InstanceLoginService({ db: options.db, client, resolver, now: options.now })
: new InstanceLoginService({ db: options.db, client, resolver });
const pendingLogins = new Map<string, Promise<void>>();
const ensureSession = async (
instanceId: string,
origin: string,
force = false,
): Promise<void> => {
const existing = sessions.sessionFor(instanceId);
if (!force && existing?.origin === origin) return;
if (existing) sessions.clear(instanceId);
const current = pendingLogins.get(instanceId);
if (current) return current;
const pending = login
.login(instanceId)
.then((result) => {
if (!result.authenticated) throw new Error('INSTANCE_AUTHENTICATION_FAILED');
})
.finally(() => pendingLogins.delete(instanceId));
pendingLogins.set(instanceId, pending);
return pending;
};
const resources = new InstanceResourceService({
instances,
sessions,
request: options.upstream.request,
ensureSession,
});
const messages = new InstanceMessageService({
instances,
sessions,
request: options.upstream.request,
ensureSession,
});
const resolver = new InstanceCredentialResolver({ db: options.db, store: options.store });
const login = options.now
? new InstanceLoginService({ db: options.db, client, resolver, now: options.now })
: new InstanceLoginService({ db: options.db, client, resolver });
const deletion = options.now
? new DeleteInstanceOperation({ db: options.db, instances, now: options.now })
: new DeleteInstanceOperation({ db: options.db, instances });
deletion.reconcileInterruptedJobs();
const resolveOperationCookie = async (
instanceId: string,
origin: string,
): Promise<string | undefined> => {
// Attach only an origin-bound in-memory session. Never invent cookies, never log them.
const existing = sessions.sessionFor(instanceId);
if (
existing &&
existing.origin === origin &&
/^simadmin_session=[^;\s,]+$/u.test(existing.cookie)
)
return existing.cookie;
// Password-protected instances need a session for reboot/restart. Best-effort Keychain login
// keeps the control plane from silently dispatching unauthenticated upstream mutations.
try {
const result = await login.login(instanceId);
if (!result.authenticated) return undefined;
} catch {
return undefined;
}
const refreshed = sessions.sessionFor(instanceId);
return refreshed &&
refreshed.origin === origin &&
/^simadmin_session=[^;\s,]+$/u.test(refreshed.cookie)
? refreshed.cookie
: undefined;
};
const secureExecution = new SecureOperationExecution({
db: options.db,
registry: secureOperationRegistry,
transport: {
request: async ({ origin, path, body, contentType }) => {
request: async ({ origin, path, body, contentType, instanceId }) => {
const cookie = await resolveOperationCookie(instanceId, origin);
if (path === '/api/network/register-auto') {
const response = await options.upstream.postNetworkRegisterAuto(origin);
const response = await options.upstream.postNetworkRegisterAuto(origin, cookie);
return { status: response.status };
}
if (path === '/api/service/restart') {
const response = await options.upstream.postServiceRestart(origin);
const response = await options.upstream.postServiceRestart(origin, cookie);
return { status: response.status };
}
if (path === '/api/system/reboot') {
if (contentType !== 'application/json' || body !== JSON.stringify({ delay_seconds: 3 }))
throw new Error('UPSTREAM_REQUEST_INVALID');
const response = await options.upstream.postSystemReboot(origin, 3);
const response = await options.upstream.postSystemReboot(origin, 3, cookie);
return { status: response.status };
}
throw new Error('UPSTREAM_REQUEST_INVALID');
@@ -110,6 +176,21 @@ export function buildControlPlaneApp(options: ControlPlaneOptions): ControlPlane
...(options.now ? { now: options.now } : {}),
});
secureExecution.reconcileInterruptedJobs();
const scheduledDispatcher = new ScheduledOperationDispatcher({
db: options.db,
operations: secureExecution,
messages,
store: options.store,
repository: scheduledTasks,
...(options.now ? { now: options.now } : {}),
});
const scheduler = new SchedulerCoordinator({
db: options.db,
repository: scheduledTasks,
dispatch: (task, targets, _run, context) =>
scheduledDispatcher.dispatch(task, targets, context),
...(options.now ? { now: options.now } : {}),
});
const auth = new ConsoleAuthService({
db: options.db,
...(options.now ? { now: options.now } : {}),
@@ -130,6 +211,11 @@ export function buildControlPlaneApp(options: ControlPlaneOptions): ControlPlane
registerOperationRoutes(app, operationCatalogRegistry, secureExecution, deletion);
registerJobRoutes(app, { jobs });
registerAuditRoutes(app, { audit });
registerAutomationRoutes(app, {
service: automation,
repository: scheduledTasks,
runNow: (taskId, actor, requestId) => scheduler.runNow(taskId, actor, requestId),
});
registerEventRoutes(app, {
journal: eventJournal,
...(options.authenticateEventStream
@@ -141,5 +227,7 @@ export function buildControlPlaneApp(options: ControlPlaneOptions): ControlPlane
Object.assign(app, {
retryPendingSecretCleanup: () => instances.retryPendingSecretCleanup(),
});
app.addHook('onClose', async () => scheduler.stop());
scheduler.start();
return app as ControlPlaneApp;
}
+2
View File
@@ -66,4 +66,6 @@ export type {
InstanceServiceOptions,
} from './application/instances/instance-service.js';
export { MIGRATIONS, migrateDatabase } from './infrastructure/database/migrations.js';
export { ScheduledTaskRepository } from './application/automation/scheduled-task-repository.js';
export { ScheduledTaskService } from './application/automation/scheduled-task-service.js';
export * as databaseSchema from './infrastructure/database/schema.js';
@@ -71,6 +71,8 @@ describe('database migrations', () => {
'job_items',
'jobs',
'operation_preparations',
'scheduled_runs',
'scheduled_tasks',
'schema_migrations',
'secret_cleanup_tasks',
'secret_references',
@@ -304,6 +304,58 @@ export const MIGRATIONS: readonly Migration[] = [
'CREATE INDEX idx_console_auth_sessions_expires_at ON console_auth_sessions(expires_at)',
],
},
{
id: 8,
name: 'scheduled-automation',
statements: [
`CREATE TABLE scheduled_tasks (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
operation_type TEXT NOT NULL CHECK (operation_type IN ('restart-service','reboot-system','send-sms')),
enabled INTEGER NOT NULL DEFAULT 1 CHECK (enabled IN (0, 1)),
version INTEGER NOT NULL DEFAULT 1 CHECK (version > 0),
cron_expression TEXT NOT NULL,
timezone TEXT NOT NULL CHECK (timezone = 'Asia/Shanghai'),
target_selector_json TEXT NOT NULL,
sms_secret_reference TEXT,
sms_recipient_count INTEGER CHECK (sms_recipient_count IS NULL OR sms_recipient_count > 0),
effective_start_at TEXT,
effective_end_at TEXT,
misfire_policy TEXT NOT NULL CHECK (misfire_policy IN ('skip','catch-up-once')),
overlap_policy TEXT NOT NULL CHECK (overlap_policy IN ('skip','queue-once')),
retry_policy_json TEXT NOT NULL,
next_due_at TEXT,
last_evaluated_at TEXT,
created_by TEXT NOT NULL,
updated_by TEXT NOT NULL,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
deleted_at TEXT,
CHECK ((operation_type = 'send-sms') = (sms_secret_reference IS NOT NULL)),
CHECK ((sms_secret_reference IS NULL) = (sms_recipient_count IS NULL))
)`,
`CREATE TABLE scheduled_runs (
id TEXT PRIMARY KEY,
scheduled_task_id TEXT NOT NULL REFERENCES scheduled_tasks(id) ON DELETE RESTRICT,
schedule_version INTEGER NOT NULL CHECK (schedule_version > 0),
task_snapshot_json TEXT NOT NULL,
due_at TEXT NOT NULL,
claimed_at TEXT NOT NULL,
started_at TEXT,
finished_at TEXT,
target_snapshot_json TEXT NOT NULL,
outcome TEXT CHECK (outcome IS NULL OR outcome IN ('succeeded','partially-succeeded','failed','skipped','no-targets','needs-attention')),
reason TEXT,
job_ids_json TEXT NOT NULL DEFAULT '[]',
trigger_source TEXT NOT NULL CHECK (trigger_source IN ('scheduled','manual')),
attempt INTEGER NOT NULL DEFAULT 1 CHECK (attempt > 0),
UNIQUE (scheduled_task_id, schedule_version, due_at, trigger_source)
)`,
'CREATE INDEX idx_scheduled_tasks_enabled_next_due ON scheduled_tasks(enabled, next_due_at) WHERE deleted_at IS NULL',
'CREATE INDEX idx_scheduled_runs_task_due ON scheduled_runs(scheduled_task_id, due_at DESC)',
'CREATE INDEX idx_scheduled_runs_outcome_finished ON scheduled_runs(outcome, finished_at DESC)',
],
},
];
const createMigrationsTable = `CREATE TABLE schema_migrations (
@@ -255,3 +255,67 @@ export const secretCleanupTasks = sqliteTable(
},
(table) => [index('idx_secret_cleanup_tasks_queued_at').on(table.queuedAt)],
);
export const scheduledTasks = sqliteTable(
'scheduled_tasks',
{
id: text('id').primaryKey(),
name: text('name').notNull(),
operationType: text('operation_type').notNull(),
enabled: integer('enabled', { mode: 'boolean' }).notNull().default(true),
version: integer('version').notNull().default(1),
cronExpression: text('cron_expression').notNull(),
timezone: text('timezone').notNull(),
targetSelectorJson: text('target_selector_json').notNull(),
smsSecretReference: text('sms_secret_reference'),
smsRecipientCount: integer('sms_recipient_count'),
effectiveStartAt: text('effective_start_at'),
effectiveEndAt: text('effective_end_at'),
misfirePolicy: text('misfire_policy').notNull(),
overlapPolicy: text('overlap_policy').notNull(),
retryPolicyJson: text('retry_policy_json').notNull(),
nextDueAt: text('next_due_at'),
lastEvaluatedAt: text('last_evaluated_at'),
createdBy: text('created_by').notNull(),
updatedBy: text('updated_by').notNull(),
createdAt: text('created_at').notNull(),
updatedAt: text('updated_at').notNull(),
deletedAt: text('deleted_at'),
},
(table) => [
check('scheduled_tasks_enabled_check', sql`${table.enabled} in (0, 1)`),
check('scheduled_tasks_version_check', sql`${table.version} > 0`),
index('idx_scheduled_tasks_enabled_next_due').on(table.enabled, table.nextDueAt),
],
);
export const scheduledRuns = sqliteTable(
'scheduled_runs',
{
id: text('id').primaryKey(),
scheduledTaskId: text('scheduled_task_id')
.notNull()
.references(() => scheduledTasks.id, { onDelete: 'restrict' }),
scheduleVersion: integer('schedule_version').notNull(),
taskSnapshotJson: text('task_snapshot_json').notNull(),
dueAt: text('due_at').notNull(),
claimedAt: text('claimed_at').notNull(),
startedAt: text('started_at'),
finishedAt: text('finished_at'),
targetSnapshotJson: text('target_snapshot_json').notNull(),
outcome: text('outcome'),
reason: text('reason'),
jobIdsJson: text('job_ids_json').notNull().default('[]'),
triggerSource: text('trigger_source').notNull(),
attempt: integer('attempt').notNull().default(1),
},
(table) => [
unique('scheduled_runs_occurrence_unique').on(
table.scheduledTaskId,
table.scheduleVersion,
table.dueAt,
table.triggerSource,
),
index('idx_scheduled_runs_task_due').on(table.scheduledTaskId, desc(table.dueAt)),
],
);
@@ -151,7 +151,23 @@ describe('MacOSKeychainSecretStore', () => {
}
});
it.each(['line\nbreak', 'line\rbreak', 'nul\0break', 'x'.repeat(4097)])(
it('stores a maximum-sized scheduled SMS payload', async () => {
const runner = new FakeRunner();
const store = new MacOSKeychainSecretStore(runner);
const payload = JSON.stringify({
recipients: Array.from(
{ length: 50 },
(_, index) => `1380013${String(index).padStart(4, '0')}`,
),
content: '字'.repeat(2_000),
});
await expect(
store.set({ instanceId: 'task-1', purpose: 'scheduled-sms', slot: 'rotation-1' }, payload),
).resolves.toMatch(/^keychain:/);
});
it.each(['line\nbreak', 'line\rbreak', 'nul\0break', 'x'.repeat(16_385)])(
'rejects non-line-safe or oversized secret input before invoking the runner',
async (value) => {
const runner = new FakeRunner();
@@ -181,7 +181,7 @@ export class MacOSKeychainSecretStore implements SecretStore {
typeof value !== 'string' ||
value.length === 0 ||
/[\0\r\n]/.test(value) ||
Buffer.byteLength(value, 'utf8') > 4096
Buffer.byteLength(value, 'utf8') > 16_384
) {
throw new SecretStoreError('INVALID_SECRET', 'Secret is not valid for Keychain storage');
}
@@ -4,9 +4,13 @@ import { SafeUpstreamGateway, type SafeUpstreamTransport } from './safe-upstream
export interface SafeControlPlaneUpstream extends ConnectionTransport {
request: UpstreamSessionClientOptions['request'];
postNetworkRegisterAuto(origin: string): Promise<{ readonly status: number }>;
postServiceRestart(origin: string): Promise<{ readonly status: number }>;
postSystemReboot(origin: string, delaySeconds: number): Promise<{ readonly status: number }>;
postNetworkRegisterAuto(origin: string, cookie?: string): Promise<{ readonly status: number }>;
postServiceRestart(origin: string, cookie?: string): Promise<{ readonly status: number }>;
postSystemReboot(
origin: string,
delaySeconds: number,
cookie?: string,
): Promise<{ readonly status: number }>;
}
export function createSafeControlPlaneUpstream(
transport: SafeUpstreamTransport,
@@ -15,8 +19,9 @@ export function createSafeControlPlaneUpstream(
return {
get: (url) => transport.get(url),
request: (request) => gateway.request(request),
postNetworkRegisterAuto: (origin) => gateway.postNetworkRegisterAuto(origin),
postServiceRestart: (origin) => gateway.postServiceRestart(origin),
postSystemReboot: (origin, delaySeconds) => gateway.postSystemReboot(origin, delaySeconds),
postNetworkRegisterAuto: (origin, cookie) => gateway.postNetworkRegisterAuto(origin, cookie),
postServiceRestart: (origin, cookie) => gateway.postServiceRestart(origin, cookie),
postSystemReboot: (origin, delaySeconds, cookie) =>
gateway.postSystemReboot(origin, delaySeconds, cookie),
};
}
@@ -65,7 +65,7 @@ describe('SafeUpstreamGateway', () => {
});
await expect(
gateway.request({
url: 'http://192.168.1.20:8080/api/auth/login',
url: 'http://example.com/api/auth/login',
method: 'POST',
headers: { 'content-type': 'application/json' },
secret: '[REDACTED]',
@@ -74,13 +74,50 @@ describe('SafeUpstreamGateway', () => {
).rejects.toThrow('UPSTREAM_INSECURE_AUTH');
await expect(
gateway.request({
url: 'http://192.168.1.20:8080/api/auth/logout',
url: 'http://example.com/api/auth/logout',
method: 'POST',
headers: { cookie: 'simadmin_session=opaque' },
}),
).rejects.toThrow('UPSTREAM_INSECURE_AUTH');
});
it('allows HTTP auth only for literal private instance hosts', async () => {
const calls: unknown[] = [];
const gateway = new SafeUpstreamGateway({
transport: {
get: async () => ({ status: 200, headers: {}, body: '' }),
post: async (url, headers, body) => {
calls.push({ url, headers, body });
return { status: 200, headers: { 'set-cookie': 'simadmin_session=opaque' }, body: '' };
},
},
});
await gateway.request({
url: 'http://192.168.1.20:8080/api/auth/login',
method: 'POST',
headers: { 'content-type': 'application/json' },
secret: '[REDACTED]',
body: '[REDACTED]',
});
await gateway.request({
url: 'http://192.168.1.20:8080/api/auth/logout',
method: 'POST',
headers: { cookie: 'simadmin_session=opaque' },
});
expect(calls).toEqual([
{
url: 'http://192.168.1.20:8080/api/auth/login',
headers: { 'content-type': 'application/json' },
body: '{"password":"[REDACTED]"}',
},
{
url: 'http://192.168.1.20:8080/api/auth/logout',
headers: { cookie: 'simadmin_session=opaque' },
body: '',
},
]);
});
it('allows audited resource GETs without a cookie for passwordless instances', async () => {
const get = vi.fn(async () => ({ status: 200, headers: {}, body: '{}' }));
const gateway = new SafeUpstreamGateway({
@@ -121,6 +158,23 @@ describe('SafeUpstreamGateway', () => {
);
});
it('dispatches an SMS at the 2,000-character automation limit', async () => {
const post = vi.fn(async () => ({ status: 200, headers: {}, body: '{}' }));
const gateway = new SafeUpstreamGateway({
transport: { get: vi.fn(), post },
});
const content = '字'.repeat(2_000);
await gateway.request({
url: 'http://192.168.1.20:8080/api/sms/send',
method: 'POST',
headers: { accept: 'application/json', 'content-type': 'application/json' },
sms: { phoneNumber: '+15550199', content },
});
expect(post).toHaveBeenCalledOnce();
});
it('rejects unallowlisted SMS queries and malformed send payloads before transport', async () => {
const get = vi.fn(async () => ({ status: 200, headers: {}, body: '{}' }));
const post = vi.fn(async () => ({ status: 200, headers: {}, body: '{}' }));
@@ -184,4 +238,35 @@ describe('SafeUpstreamGateway', () => {
},
]);
});
it('forwards only a canonical origin-bound simadmin_session cookie on restart ops', async () => {
const calls: unknown[] = [];
const gateway = new SafeUpstreamGateway({
transport: {
get: async () => ({ status: 200, headers: {}, body: '' }),
post: async (url, headers, body) => {
calls.push({ url, headers, body });
return { status: 204, headers: {}, body: '' };
},
},
});
const cookie = 'simadmin_session=opaque-token';
await gateway.postServiceRestart('http://192.168.1.20:8080', cookie);
await gateway.postSystemReboot('http://192.168.1.20:8080', 3, cookie);
expect(calls).toEqual([
{
url: 'http://192.168.1.20:8080/api/service/restart',
headers: { cookie },
body: '',
},
{
url: 'http://192.168.1.20:8080/api/system/reboot',
headers: { 'content-type': 'application/json', cookie },
body: '{"delay_seconds":3}',
},
]);
await expect(
gateway.postServiceRestart('http://192.168.1.20:8080', 'simadmin_session=bad; extra'),
).rejects.toMatchObject({ code: 'UPSTREAM_REQUEST_INVALID', dispatched: false });
});
});
@@ -14,6 +14,25 @@ export interface SafeUpstreamTransport {
body: string,
): Promise<TransportResponse>;
}
/** Literal private hosts only — hostnames that still need DNS stay HTTPS for auth secrets. */
const isLiteralPrivateHost = (hostname: string): boolean => {
const host = hostname.replace(/^\[|\]$/g, '').toLowerCase();
if (host.includes(':')) return host.startsWith('fc') || host.startsWith('fd');
const parts = host.split('.').map(Number);
if (parts.length !== 4 || parts.some((part) => !Number.isInteger(part) || part < 0 || part > 255))
return false;
const [a, b, c] = parts;
if (a === undefined || b === undefined || c === undefined) return false;
return (
a === 10 ||
(a === 172 && b >= 16 && b <= 31) ||
(a === 192 && b === 168) ||
(a === 100 && b >= 64 && b <= 127) ||
(a === 192 && b === 0 && c === 0)
);
};
export class SafeUpstreamGateway {
constructor(private readonly options: { readonly transport: SafeUpstreamTransport }) {}
private assertOrigin(origin: string): string {
@@ -35,11 +54,24 @@ export class SafeUpstreamGateway {
return parsed.origin;
}
private async postZeroBody(origin: string, path: string): Promise<UpstreamResponse> {
private sessionHeaders(
cookie: string | undefined,
extra: Readonly<Record<string, string>> = {},
): Readonly<Record<string, string>> {
if (cookie === undefined) return extra;
if (!/^simadmin_session=[^;\s,]+$/u.test(cookie)) throw this.notDispatched();
return { ...extra, cookie };
}
private async postZeroBody(
origin: string,
path: string,
cookie?: string,
): Promise<UpstreamResponse> {
const base = this.assertOrigin(origin);
const url = `${base}${path}`;
try {
return await this.options.transport.post(url, {}, '');
return await this.options.transport.post(url, this.sessionHeaders(cookie), '');
} catch (error) {
if (
error instanceof UpstreamError &&
@@ -50,22 +82,26 @@ export class SafeUpstreamGateway {
}
}
async postNetworkRegisterAuto(origin: string): Promise<UpstreamResponse> {
return this.postZeroBody(origin, '/api/network/register-auto');
async postNetworkRegisterAuto(origin: string, cookie?: string): Promise<UpstreamResponse> {
return this.postZeroBody(origin, '/api/network/register-auto', cookie);
}
async postServiceRestart(origin: string): Promise<UpstreamResponse> {
return this.postZeroBody(origin, '/api/service/restart');
async postServiceRestart(origin: string, cookie?: string): Promise<UpstreamResponse> {
return this.postZeroBody(origin, '/api/service/restart', cookie);
}
async postSystemReboot(origin: string, delaySeconds: number): Promise<UpstreamResponse> {
async postSystemReboot(
origin: string,
delaySeconds: number,
cookie?: string,
): Promise<UpstreamResponse> {
if (delaySeconds !== 3) throw this.notDispatched();
const base = this.assertOrigin(origin);
const url = `${base}/api/system/reboot`;
try {
return await this.options.transport.post(
url,
{ 'content-type': 'application/json' },
this.sessionHeaders(cookie, { 'content-type': 'application/json' }),
JSON.stringify({ delay_seconds: 3 }),
);
} catch (error) {
@@ -92,7 +128,10 @@ export class SafeUpstreamGateway {
request.body !== undefined ||
request.sms !== undefined ||
(headerKeys !== 'accept' && headerKeys !== 'accept,cookie') ||
(url.pathname !== '/api/stats' && url.pathname !== '/api/sim' && !smsQuery) ||
(url.pathname !== '/api/stats' &&
url.pathname !== '/api/sim' &&
url.pathname !== '/api/health' &&
!smsQuery) ||
(!smsList && url.search) ||
url.hash ||
url.username ||
@@ -114,8 +153,8 @@ export class SafeUpstreamGateway {
(headerKeys !== 'accept,content-type' && headerKeys !== 'accept,content-type,cookie') ||
!/^\+?[0-9][0-9 ()-]{2,31}$/u.test(sms.phoneNumber) ||
sms.content.length < 1 ||
sms.content.length > 1600 ||
Buffer.byteLength(sms.content, 'utf8') > 6400 ||
sms.content.length > 2000 ||
Buffer.byteLength(sms.content, 'utf8') > 8000 ||
/[\u0000-\u0008\u000b\u000c\u000e-\u001f\u007f]/u.test(sms.content) ||
url.search ||
url.hash ||
@@ -129,7 +168,9 @@ export class SafeUpstreamGateway {
JSON.stringify({ phone_number: sms.phoneNumber, content: sms.content }),
);
}
if (url.protocol !== 'https:') throw new UpstreamError('UPSTREAM_INSECURE_AUTH');
// Auth secrets may travel over HTTPS anywhere, or over HTTP only to literal private hosts.
// Public/cleartext auth is still rejected here; SSRF private-only dial remains in transport.
if (!this.allowsAuthProtocol(url)) throw new UpstreamError('UPSTREAM_INSECURE_AUTH');
if (request.method !== 'POST') throw new UpstreamError('UPSTREAM_REQUEST_INVALID');
if (request.url.endsWith('/api/auth/login')) {
if (typeof request.secret !== 'string' || request.body !== '[REDACTED]')
@@ -147,6 +188,12 @@ export class SafeUpstreamGateway {
}
throw new UpstreamError('UPSTREAM_REQUEST_INVALID');
}
private allowsAuthProtocol(url: URL): boolean {
if (url.protocol === 'https:') return true;
if (url.protocol !== 'http:') return false;
return isLiteralPrivateHost(url.hostname);
}
private notDispatched(): OperationNotDispatchedError {
return new OperationNotDispatchedError('UPSTREAM_REQUEST_INVALID');
}
@@ -0,0 +1,185 @@
import Database from 'better-sqlite3';
import { describe, expect, it } from 'vitest';
import { buildApp } from '../../app.js';
import { ScheduledTaskRepository } from '../../application/automation/scheduled-task-repository.js';
import { ScheduledTaskService } from '../../application/automation/scheduled-task-service.js';
import { migrateDatabase } from '../../infrastructure/database/migrations.js';
import type { SecretKey, SecretStore } from '../../infrastructure/secrets/secret-store.js';
import { registerAutomationRoutes } from './automation-routes.js';
class MemorySecrets implements SecretStore {
readonly values = new Map<string, string>();
async set(key: SecretKey, value: string): Promise<string> {
const reference = `memory://${key.instanceId}/${key.purpose}`;
this.values.set(reference, value);
return reference;
}
async get(reference: string) {
return this.values.get(reference);
}
async delete(reference: string) {
return this.values.delete(reference);
}
}
function fixture() {
const db = new Database(':memory:');
db.pragma('foreign_keys = ON');
migrateDatabase(db);
const repository = new ScheduledTaskRepository(db);
let id = 0;
const service = new ScheduledTaskService({
repository,
store: new MemorySecrets(),
now: () => new Date('2026-07-30T00:00:00.000Z'),
id: () => `task-${++id}`,
});
const app = buildApp({
registerRoutes: (scope) => registerAutomationRoutes(scope, { service, repository }),
});
return { app, db };
}
const restart = {
name: 'Morning restart',
operationType: 'restart-service',
cronExpression: '0 9 * * *',
targetSelector: { mode: 'fixed', instanceIds: ['instance-a'] },
};
describe('automation HTTP routes', () => {
it('previews Beijing-time Cron and creates a redacted schedule', async () => {
const { app, db } = fixture();
const preview = await app.inject({
method: 'POST',
url: '/api/v1/automation/cron/preview',
payload: { cronExpression: '0 9 * * *', count: 2 },
});
expect(preview.statusCode).toBe(200);
expect(preview.json()).toEqual({
timezone: 'Asia/Shanghai',
occurrences: ['2026-07-30T01:00:00.000Z', '2026-07-31T01:00:00.000Z'],
});
const created = await app.inject({
method: 'POST',
url: '/api/v1/automation/schedules',
payload: restart,
});
expect(created.statusCode).toBe(201);
expect(created.headers.etag).toBe('"version-1"');
expect(created.json()).toMatchObject({ id: 'task-1', timezone: 'Asia/Shanghai' });
expect(created.body).not.toMatch(/confirmationToken|smsSecretReference/i);
await app.close();
db.close();
});
it('requires optimistic version headers for state changes and preserves run history on delete', async () => {
const { app, db } = fixture();
await app.inject({ method: 'POST', url: '/api/v1/automation/schedules', payload: restart });
const missing = await app.inject({
method: 'PATCH',
url: '/api/v1/automation/schedules/task-1/state',
payload: { enabled: false },
});
expect(missing.statusCode).toBe(428);
const paused = await app.inject({
method: 'PATCH',
url: '/api/v1/automation/schedules/task-1/state',
headers: { 'if-match': '"version-1"' },
payload: { enabled: false },
});
expect(paused.statusCode).toBe(200);
expect(paused.json()).toMatchObject({ enabled: false, version: 2 });
const removed = await app.inject({
method: 'DELETE',
url: '/api/v1/automation/schedules/task-1',
headers: { 'if-match': '"version-2"' },
});
expect(removed.statusCode).toBe(204);
expect(
(await app.inject({ method: 'GET', url: '/api/v1/automation/schedules' })).json(),
).toMatchObject({ items: [] });
await app.close();
db.close();
});
it('rejects alternate timezone and unknown preview keys with sanitized problems', async () => {
const { app, db } = fixture();
const invalid = await app.inject({
method: 'POST',
url: '/api/v1/automation/schedules',
payload: { ...restart, timezone: 'UTC' },
});
expect(invalid.statusCode).toBe(400);
expect(invalid.json()).toMatchObject({ code: 'AUTOMATION_VALIDATION_FAILED' });
expect(invalid.body).not.toContain('stack');
await app.close();
db.close();
});
it('updates and duplicates schedules with optimistic version checks', async () => {
const { app, db } = fixture();
await app.inject({ method: 'POST', url: '/api/v1/automation/schedules', payload: restart });
const missingVersion = await app.inject({
method: 'PATCH',
url: '/api/v1/automation/schedules/task-1',
payload: { name: 'Updated restart' },
});
expect(missingVersion.statusCode).toBe(428);
const updated = await app.inject({
method: 'PATCH',
url: '/api/v1/automation/schedules/task-1',
headers: { 'if-match': '"version-1"' },
payload: { name: 'Updated restart', effectiveEndAt: '2026-08-30T00:00:00.000Z' },
});
expect(updated.statusCode).toBe(200);
expect(updated.headers.etag).toBe('"version-2"');
expect(updated.json()).toMatchObject({ name: 'Updated restart', version: 2 });
const duplicated = await app.inject({
method: 'POST',
url: '/api/v1/automation/schedules/task-1/duplicate',
headers: { 'if-match': '"version-2"' },
});
expect(duplicated.statusCode).toBe(201);
expect(duplicated.json()).toMatchObject({
id: 'task-2',
name: 'Updated restart copy',
enabled: false,
});
await app.close();
db.close();
});
it('bounds and validates schedule pagination', async () => {
const { app, db } = fixture();
for (const name of ['One', 'Two', 'Three'])
await app.inject({
method: 'POST',
url: '/api/v1/automation/schedules',
payload: { ...restart, name },
});
const page = await app.inject({
method: 'GET',
url: '/api/v1/automation/schedules?page=2&pageSize=1',
});
expect(page.statusCode).toBe(200);
expect(page.json()).toMatchObject({
items: [{ id: 'task-2' }],
page: { page: 2, pageSize: 1, total: 3 },
});
const invalid = await app.inject({
method: 'GET',
url: '/api/v1/automation/schedules?pageSize=101',
});
expect(invalid.statusCode).toBe(400);
expect(invalid.json()).toMatchObject({ code: 'AUTOMATION_VALIDATION_FAILED' });
await app.close();
db.close();
});
});
@@ -0,0 +1,239 @@
import { AUTOMATION_TIMEZONE } from '@multi-simadmin/contracts';
import type { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify';
import { ScheduledTaskRepository } from '../../application/automation/scheduled-task-repository.js';
import { ScheduledTaskService } from '../../application/automation/scheduled-task-service.js';
export interface AutomationRoutesOptions {
readonly service: ScheduledTaskService;
readonly repository: ScheduledTaskRepository;
readonly runNow?: (taskId: string, actor: string, requestId: string) => Promise<unknown>;
}
function problem(
request: FastifyRequest,
reply: FastifyReply,
status: number,
code: string,
detail: string,
) {
return reply
.code(status)
.type('application/problem+json')
.send({
type: 'about:blank',
title: status === 404 ? 'Not Found' : status === 409 ? 'Conflict' : 'Bad Request',
status,
code,
detail,
requestId: request.id,
});
}
function version(request: FastifyRequest, reply: FastifyReply): number | undefined {
const value = request.headers['if-match'];
const match = typeof value === 'string' ? /^"version-([1-9]\d*)"$/.exec(value) : null;
if (!match) {
problem(request, reply, 428, 'VERSION_REQUIRED', 'A current schedule version is required.');
return undefined;
}
return Number(match[1]);
}
function object(value: unknown): Record<string, unknown> {
if (typeof value !== 'object' || value === null || Array.isArray(value))
throw new TypeError('Request body must be an object');
return value as Record<string, unknown>;
}
function pagination(
value: unknown,
extraKeys: readonly string[] = [],
): { readonly page: number; readonly pageSize: number; readonly query: Record<string, unknown> } {
const query = object(value);
const allowed = new Set(['page', 'pageSize', ...extraKeys]);
if (Object.keys(query).some((key) => !allowed.has(key)))
throw new TypeError('Unknown automation list query');
const integer = (key: 'page' | 'pageSize', fallback: number, maximum: number): number => {
const raw = query[key];
if (raw === undefined) return fallback;
if (typeof raw !== 'string' || !/^[1-9]\d*$/.test(raw))
throw new TypeError(`${key} is invalid`);
const parsed = Number(raw);
if (!Number.isSafeInteger(parsed) || parsed > maximum) throw new TypeError(`${key} is invalid`);
return parsed;
};
const pageSize = integer('pageSize', 25, 100);
const page = integer('page', 1, Math.floor(Number.MAX_SAFE_INTEGER / pageSize) + 1);
return { page, pageSize, query };
}
function pageItems<T>(items: readonly T[], page: number, pageSize: number) {
const offset = (page - 1) * pageSize;
return {
items: items.slice(offset, offset + pageSize),
page: { page, pageSize, total: items.length },
};
}
async function action(
request: FastifyRequest,
reply: FastifyReply,
operation: () => unknown | Promise<unknown>,
) {
try {
return await operation();
} catch (error) {
if (error instanceof TypeError)
return problem(
request,
reply,
400,
'AUTOMATION_VALIDATION_FAILED',
'The automation request is invalid.',
);
if (error instanceof Error && /not found|version changed/i.test(error.message))
return problem(
request,
reply,
409,
'SCHEDULE_VERSION_CONFLICT',
'The schedule changed or no longer exists.',
);
throw error;
}
}
const etag = (version: number) => `"version-${version}"`;
export function registerAutomationRoutes(
app: FastifyInstance,
options: AutomationRoutesOptions,
): void {
app.post('/api/v1/automation/cron/preview', async (request, reply) =>
action(request, reply, () => {
const body = object(request.body);
if (Object.keys(body).some((key) => key !== 'cronExpression' && key !== 'count'))
throw new TypeError('Unknown preview field');
if (typeof body.cronExpression !== 'string')
throw new TypeError('cronExpression is required');
const count = body.count === undefined ? 5 : body.count;
if (!Number.isSafeInteger(count)) throw new TypeError('count is invalid');
return {
timezone: AUTOMATION_TIMEZONE,
occurrences: options.service.preview(body.cronExpression, count as number),
};
}),
);
app.get('/api/v1/automation/schedules', async (request, reply) =>
action(request, reply, () => {
const { page, pageSize } = pagination(request.query);
return pageItems(options.service.list(), page, pageSize);
}),
);
app.get('/api/v1/automation/schedules/:taskId', async (request, reply) => {
const task = options.service.get((request.params as { taskId: string }).taskId);
if (!task)
return problem(request, reply, 404, 'SCHEDULE_NOT_FOUND', 'The schedule does not exist.');
return reply.header('ETag', etag(task.version)).send(task);
});
app.post('/api/v1/automation/schedules', async (request, reply) =>
action(request, reply, async () => {
const task = await options.service.create('console-operator', request.body);
return reply.code(201).header('ETag', etag(task.version)).send(task);
}),
);
app.patch('/api/v1/automation/schedules/:taskId', async (request, reply) => {
const currentVersion = version(request, reply);
if (currentVersion === undefined) return reply;
return action(request, reply, async () => {
const task = await options.service.update(
'console-operator',
(request.params as { taskId: string }).taskId,
currentVersion,
request.body,
);
return reply.header('ETag', etag(task.version)).send(task);
});
});
app.post('/api/v1/automation/schedules/:taskId/duplicate', async (request, reply) => {
const currentVersion = version(request, reply);
if (currentVersion === undefined) return reply;
return action(request, reply, async () => {
const task = await options.service.duplicate(
'console-operator',
(request.params as { taskId: string }).taskId,
currentVersion,
);
return reply.code(201).header('ETag', etag(task.version)).send(task);
});
});
app.patch('/api/v1/automation/schedules/:taskId/state', async (request, reply) => {
const currentVersion = version(request, reply);
if (currentVersion === undefined) return reply;
return action(request, reply, () => {
const body = object(request.body);
if (Object.keys(body).length !== 1 || typeof body.enabled !== 'boolean')
throw new TypeError('enabled is required');
const task = options.service.setEnabled(
'console-operator',
(request.params as { taskId: string }).taskId,
currentVersion,
body.enabled,
);
return reply.header('ETag', etag(task.version)).send(task);
});
});
app.delete('/api/v1/automation/schedules/:taskId', async (request, reply) => {
const currentVersion = version(request, reply);
if (currentVersion === undefined) return reply;
return action(request, reply, async () => {
await options.service.remove(
'console-operator',
(request.params as { taskId: string }).taskId,
currentVersion,
);
return reply.code(204).send();
});
});
app.get('/api/v1/automation/runs', async (request, reply) =>
action(request, reply, () => {
const { page, pageSize, query } = pagination(request.query, ['scheduledTaskId']);
const scheduledTaskId =
typeof query.scheduledTaskId === 'string' && query.scheduledTaskId
? query.scheduledTaskId
: undefined;
if (query.scheduledTaskId !== undefined && !scheduledTaskId)
throw new TypeError('scheduledTaskId is invalid');
return pageItems(options.repository.listRuns(scheduledTaskId), page, pageSize);
}),
);
app.post('/api/v1/automation/schedules/:taskId/run', async (request, reply) => {
const runNow = options.runNow;
if (!runNow)
return problem(
request,
reply,
409,
'SCHEDULER_UNAVAILABLE',
'The scheduler is not available.',
);
return action(request, reply, async () => {
const run = await runNow(
(request.params as { taskId: string }).taskId,
'console-operator',
request.id,
);
return reply.code(202).send(run);
});
});
}
@@ -400,7 +400,7 @@ export function registerInstanceRoutes(app: FastifyInstance, options: InstanceRo
required: ['phoneNumber', 'content'],
properties: {
phoneNumber: { type: 'string', minLength: 3, maxLength: 32 },
content: { type: 'string', minLength: 1, maxLength: 1600 },
content: { type: 'string', minLength: 1, maxLength: 2000 },
},
},
},
@@ -31,8 +31,8 @@ async function fixtureOptions(): Promise<ProductionControlPlaneOptions> {
get: async () => ({ status: 200, headers: {}, body: '' }),
request: async () => ({ status: 200, headers: {}, body: '' }),
postNetworkRegisterAuto: async () => ({ status: 200 }),
postServiceRestart: async () => ({ status: 200 }),
postSystemReboot: async () => ({ status: 200 }),
postServiceRestart: async () => ({ status: 200 }),
postSystemReboot: async () => ({ status: 200 }),
},
keychainMetadataCheck: async () => true,
};