import type { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify'; import type { Instance } from '@multi-simadmin/contracts'; import type { InstanceService } from '../../application/instances/instance-service.js'; import type { InstanceResourceService } from '../../application/resources/instance-resource-service.js'; import type { ConnectionState, ConnectionProbe, } from '../../application/connections/connection-probe.js'; import { MessageServiceError, type DeleteSmsMessageRequest, } from '../../application/messages/instance-message-service.js'; import type { HubMessageService } from '../../application/messages/hub-message-service.js'; import { type SmsOutboxItem, type SmsOutboxService, type SmsOutboxStatus, } from '../../application/messages/sms-outbox-service.js'; import { NotificationServiceError, type InstanceNotificationService, } from '../../application/notifications/instance-notification-service.js'; import type { DeviceIdentity, DeviceIdentityService, IdentityReason, } from '../../application/identity/device-identity-service.js'; export interface FleetMessageDevice { readonly id: string; readonly name: string; readonly availability: 'online' | 'unavailable'; } export interface FleetMessage { readonly id: string; readonly instanceId: string; readonly instanceName: string; readonly direction: string; readonly phoneNumber: string; readonly content: string; readonly timestamp: string; readonly status: string; readonly transport: string; } export interface FleetMessagesResponse { readonly messages: readonly FleetMessage[]; readonly devices: readonly FleetMessageDevice[]; readonly total: number; } export interface FleetMessageConversation { readonly instanceId: string; readonly instanceName: string; readonly phoneNumber: string; readonly messageCount: number; readonly incomingCount: number; readonly lastMessage: FleetMessage; } export interface FleetMessageConversationsResponse { readonly conversations: readonly FleetMessageConversation[]; readonly total: number; readonly stats: { readonly incoming: number; readonly outgoing: number; readonly total: number }; } export interface FleetRoutesOptions { readonly instances: InstanceService; readonly resources: InstanceResourceService; readonly messages: HubMessageService; readonly notifications: InstanceNotificationService; /** Reads the heartbeat journal; the overview never probes live so the page stays cheap. */ readonly connections?: ConnectionProbe; /** Offline send queue; absent means sends fail fast instead of waiting for the device. */ readonly outbox?: SmsOutboxService; /** Identity guard; absent means the fleet list has no opinion about hardware drift. */ readonly identities?: DeviceIdentityService; } export interface FleetIdentitySummary { readonly tracked: boolean; readonly status: 'confirmed' | 'pending'; readonly reasons: readonly IdentityReason[]; } /** One queued send, with the node label the queue table itself does not store. */ export type FleetOutboxItem = SmsOutboxItem & { readonly instanceName: string }; function parseOutboxQuery(query: unknown): { readonly status: SmsOutboxStatus | 'open' | 'all'; readonly instanceId?: string; readonly limit: number; readonly offset: number; } { const value = record(query) ?? {}; if (Object.keys(value).some((key) => !['status', 'instanceId', 'limit', 'offset'].includes(key))) throw new MessageServiceError('VALIDATION_FAILED'); const status = value.status; if (status !== undefined && typeof status !== 'string') throw new MessageServiceError('VALIDATION_FAILED'); const resolved = (status ?? 'open') as SmsOutboxStatus | 'open' | 'all'; if (!OUTBOX_STATUSES.includes(resolved)) throw new MessageServiceError('VALIDATION_FAILED'); const instanceId = value.instanceId; if ( instanceId !== undefined && (typeof instanceId !== 'string' || instanceId.length > 256 || instanceId.trim() === '') ) throw new MessageServiceError('VALIDATION_FAILED'); const integer = (key: 'limit' | 'offset', fallback: number, maximum: number): number => { const raw = value[key]; if (raw === undefined) return fallback; if (typeof raw !== 'string' || !/^\d+$/u.test(raw)) throw new MessageServiceError('VALIDATION_FAILED'); const parsed = Number(raw); if (!Number.isSafeInteger(parsed) || parsed < 0 || parsed > maximum) throw new MessageServiceError('VALIDATION_FAILED'); return parsed; }; return { status: resolved, ...(typeof instanceId === 'string' && instanceId.trim() !== '' ? { instanceId: instanceId.trim() } : {}), limit: integer('limit', 25, MAX_FLEET_OUTBOX_LIMIT), offset: integer('offset', 0, MAX_FLEET_MESSAGE_OFFSET), }; } function outboxQueueId(value: unknown): string { const id = typeof value === 'string' ? value : ''; if (!/^[A-Za-z0-9_.-]{1,64}$/u.test(id)) throw new MessageServiceError('VALIDATION_FAILED'); return id; } const OUTBOX_STATUSES: readonly (SmsOutboxStatus | 'open' | 'all')[] = [ 'open', 'all', 'queued', 'sending', 'sent', 'failed', 'cancelled', ]; /** * Heartbeat view of one device. `probed: false` means the control plane has never reached it, * which the fleet table shows as 未知 rather than a misleading 离线. */ export interface FleetConnectionSummary { readonly probed: boolean; readonly reachable: boolean; readonly authenticated: boolean; readonly checkedAt: string | null; } const PAGE_SIZE = 100; const MAX_FLEET_DEVICES = 200; const MAX_FLEET_MESSAGE_LIMIT = 100; const MAX_FLEET_CONVERSATION_LIMIT = 200; const MAX_FLEET_MESSAGE_OFFSET = 1000; const MAX_FLEET_OUTBOX_LIMIT = 100; const EMPTY_OUTBOX_SUMMARY = Object.freeze({ queued: 0, sending: 0, failed: 0, sent: 0 }); interface QueryMessagesInput { readonly limit: number; readonly offset: number; readonly search?: string; readonly instanceId?: string; readonly phoneNumber?: string; } /** Thread-level direction filter, accepted only by the conversation endpoint. */ export interface QueryConversationsInput extends QueryMessagesInput { readonly direction?: 'incoming' | 'outgoing'; } async function listInstances(instances: InstanceService): Promise { const result: Instance[] = []; let page = 1; while (result.length < MAX_FLEET_DEVICES) { const current = await instances.list({ page, pageSize: PAGE_SIZE }); result.push(...current.items.slice(0, MAX_FLEET_DEVICES - result.length)); if (current.items.length < PAGE_SIZE) break; page += 1; } return result; } function connectionSummary(state: ConnectionState | undefined): FleetConnectionSummary { if (!state) return { probed: false, reachable: false, authenticated: false, checkedAt: null }; return { probed: true, reachable: state.reachable, authenticated: state.authenticated, checkedAt: state.checkedAt, }; } function identitySummary(identity: DeviceIdentity | undefined): FleetIdentitySummary { if (!identity) return { tracked: false, status: 'confirmed', reasons: [] }; return { tracked: true, status: identity.status, reasons: identity.reasons }; } function record(value: unknown): Record | undefined { if (!value || typeof value !== 'object' || Array.isArray(value)) return undefined; return value as Record; } function parseMessagesQuery( query: unknown, maximumLimit = MAX_FLEET_MESSAGE_LIMIT, allowDirection = false, ): QueryConversationsInput { const value = record(query) ?? {}; if ( Object.keys(value).some( (key) => !['limit', 'offset', 'search', 'instanceId', 'phoneNumber'].includes(key) && !(allowDirection && key === 'direction'), ) ) throw new MessageServiceError('VALIDATION_FAILED'); const integer = (key: 'limit' | 'offset', fallback: number, maximum: number): number => { const raw = value[key]; if (raw === undefined) return fallback; if (typeof raw !== 'string' || !/^\d+$/u.test(raw)) { throw new MessageServiceError('VALIDATION_FAILED'); } const parsed = Number(raw); if (!Number.isSafeInteger(parsed) || parsed < 0 || parsed > maximum) throw new MessageServiceError('VALIDATION_FAILED'); return parsed; }; const search = value.search; if (search !== undefined && (typeof search !== 'string' || search.length > 200)) throw new MessageServiceError('VALIDATION_FAILED'); const instanceId = value.instanceId; if ( instanceId !== undefined && (typeof instanceId !== 'string' || instanceId.length > 256 || instanceId.trim() === '') ) throw new MessageServiceError('VALIDATION_FAILED'); const phoneNumber = value.phoneNumber; if ( phoneNumber !== undefined && (typeof phoneNumber !== 'string' || phoneNumber.length > 32 || phoneNumber.trim() === '') ) throw new MessageServiceError('VALIDATION_FAILED'); const direction = value.direction; if ( direction !== undefined && (!allowDirection || (direction !== 'incoming' && direction !== 'outgoing')) ) throw new MessageServiceError('VALIDATION_FAILED'); return { limit: integer('limit', 24, maximumLimit), offset: integer('offset', 0, MAX_FLEET_MESSAGE_OFFSET), ...(typeof search === 'string' && search.trim() !== '' ? { search: search.trim() } : {}), ...(typeof instanceId === 'string' && instanceId.trim() !== '' ? { instanceId: instanceId.trim() } : {}), ...(typeof phoneNumber === 'string' && phoneNumber.trim() !== '' ? { phoneNumber: phoneNumber.trim() } : {}), ...(direction === 'incoming' || direction === 'outgoing' ? { direction } : {}), }; } function parseMessageDeleteItems(value: unknown): readonly DeleteSmsMessageRequest[] { const body = record(value); const items = body?.items; if (!Array.isArray(items) || items.length < 1 || items.length > 500) throw new MessageServiceError('VALIDATION_FAILED'); return items.map((item) => { const current = record(item); const instanceId = typeof current?.instanceId === 'string' ? current.instanceId : ''; const id = typeof current?.id === 'string' ? current.id : typeof current?.id === 'number' && Number.isSafeInteger(current.id) && current.id >= 0 ? String(current.id) : ''; if (!instanceId || !id) throw new MessageServiceError('VALIDATION_FAILED'); return { instanceId, id }; }); } async function mapWithConcurrency( items: readonly Item[], concurrency: number, mapper: (item: Item, index: number) => Promise, ): Promise { const results: Result[] = new Array(items.length); let next = 0; const workers = Array.from({ length: Math.min(concurrency, items.length) }, async () => { while (next < items.length) { const index = next; next += 1; results[index] = await mapper(items[index]!, index); } }); await Promise.all(workers); return results; } function wrapMessageAction( handler: (request: FastifyRequest) => Promise, ): (request: FastifyRequest, reply: FastifyReply) => Promise { return async (request, reply) => { try { return await handler(request); } catch (error) { if (error instanceof MessageServiceError) { const status = error.code === 'NOT_FOUND' ? 404 : error.code === 'VALIDATION_FAILED' ? 400 : 502; return reply .code(status) .type('application/problem+json') .send({ type: 'about:blank', title: status === 400 ? 'Bad Request' : status === 404 ? 'Not Found' : 'Bad Gateway', status, code: error.code, detail: 'The requested Fleet message operation could not be completed.', requestId: request.id, }); } throw error; } }; } interface FleetNotificationQueueItemParams { readonly instanceId: string; readonly queueId: string; } function wrapNotificationAction( handler: ( request: FastifyRequest<{ readonly Params: FleetNotificationQueueItemParams }>, ) => Promise, ): (request: FastifyRequest, reply: FastifyReply) => Promise { return async (request, reply) => { try { return await handler( request as FastifyRequest<{ readonly Params: FleetNotificationQueueItemParams }>, ); } catch (error) { if (error instanceof NotificationServiceError) { const status = error.code === 'NOT_FOUND' ? 404 : error.code === 'VALIDATION_FAILED' ? 400 : 502; return reply .code(status) .type('application/problem+json') .send({ type: 'about:blank', title: status === 400 ? 'Bad Request' : status === 404 ? 'Not Found' : 'Bad Gateway', status, code: error.code, detail: 'The requested Fleet notification queue operation could not be completed.', requestId: request.id, }); } throw error; } }; } export function registerFleetRoutes(app: FastifyInstance, options: FleetRoutesOptions): void { app.get('/api/v1/fleet/overview', async () => { const instances = await listInstances(options.instances); const reachability = options.connections?.reachability(); const items = await mapWithConcurrency(instances, 6, async (instance) => ({ ...instance, connection: connectionSummary(reachability?.get(instance.id)), resources: await options.resources.get(instance.id), identity: identitySummary(options.identities?.get(instance.id)), })); return { items }; }); app.get('/api/v1/fleet/notifications', async () => options.notifications.summarize()); app.post('/api/v1/fleet/notifications/queue/retry-all', async () => options.notifications.retryAllQueue(), ); app.post( '/api/v1/fleet/notifications/queue/:instanceId/items/:queueId/retry', wrapNotificationAction(async (request) => options.notifications.retryQueueItem(request.params.instanceId, request.params.queueId), ), ); app.delete( '/api/v1/fleet/notifications/queue/:instanceId/items/:queueId', wrapNotificationAction(async (request) => options.notifications.deleteQueueItem(request.params.instanceId, request.params.queueId), ), ); app.get( '/api/v1/fleet/messages', wrapMessageAction(async (request) => options.messages.snapshot(parseMessagesQuery(request.query)), ), ); app.get( '/api/v1/fleet/messages/conversations', wrapMessageAction(async (request) => { const query = parseMessagesQuery(request.query, MAX_FLEET_CONVERSATION_LIMIT, true); await options.messages.refresh(); const page = options.messages.conversations(query); return { conversations: page.items.map((item) => ({ instanceId: item.instanceId, instanceName: item.instanceName, phoneNumber: item.phoneNumber, messageCount: item.messageCount, incomingCount: item.incomingCount, lastMessage: { id: item.lastMessage.id, instanceId: item.lastMessage.instanceId, instanceName: item.lastMessage.instanceName, direction: item.lastMessage.direction, phoneNumber: item.lastMessage.phoneNumber, content: item.lastMessage.content, timestamp: item.lastMessage.timestamp, status: item.lastMessage.status, transport: item.lastMessage.transport, }, })), total: page.totalCount, stats: page.stats, } satisfies FleetMessageConversationsResponse; }), ); app.post( '/api/v1/fleet/messages/sync', { bodyLimit: 4_096, schema: { body: { type: 'object', additionalProperties: false, properties: { instanceId: { type: 'string', minLength: 1, maxLength: 256 }, }, }, }, }, wrapMessageAction(async (request) => { const body = record(request.body) ?? {}; const instanceId = typeof body.instanceId === 'string' ? body.instanceId : undefined; return instanceId ? { synced: await options.messages.syncDevice(instanceId), instanceId } : await options.messages.syncAll(); }), ); app.post( '/api/v1/fleet/messages/send', { bodyLimit: 8_192, schema: { body: { type: 'object', additionalProperties: false, required: ['instanceId', 'phoneNumber', 'content'], properties: { instanceId: { type: 'string', minLength: 1, maxLength: 256 }, phoneNumber: { type: 'string', minLength: 3, maxLength: 32 }, content: { type: 'string', minLength: 1, maxLength: 2000 }, }, }, }, }, wrapMessageAction(async (request) => { const value = record(request.body); const instanceId = typeof value?.instanceId === 'string' ? value.instanceId : ''; const phoneNumber = typeof value?.phoneNumber === 'string' ? value.phoneNumber : ''; const content = typeof value?.content === 'string' ? value.content : ''; if (!options.outbox) { await options.messages.send(instanceId, { phoneNumber, content }); return { sent: true, queued: false, instanceId }; } // The queue owns a foreign key onto instances, so an unknown node stays a 404. if (!(await options.instances.get(instanceId))) throw new MessageServiceError('NOT_FOUND'); const result = await options.outbox.submit(instanceId, { phoneNumber, content }); return { sent: result.status === 'sent', queued: result.status === 'queued', instanceId, ...(result.item ? { queueId: result.item.id } : {}), }; }), ); app.post( '/api/v1/fleet/messages/delete', { bodyLimit: 32_768, schema: { body: { type: 'object', additionalProperties: false, required: ['items'], properties: { items: { type: 'array', minItems: 1, maxItems: 500, items: { type: 'object', additionalProperties: false, required: ['instanceId', 'id'], properties: { instanceId: { type: 'string', minLength: 1, maxLength: 256 }, id: { anyOf: [ { type: 'string', minLength: 1, maxLength: 64 }, { type: 'integer', minimum: 0 }, ], }, }, }, }, }, }, }, }, wrapMessageAction(async (request) => options.messages.deleteMany(parseMessageDeleteItems(request.body)), ), ); app.get( '/api/v1/fleet/messages/outbox', wrapMessageAction(async (request) => { if (!options.outbox) return { items: [], total: 0, summary: EMPTY_OUTBOX_SUMMARY }; const query = parseOutboxQuery(request.query); const page = options.outbox.list(query); const names = new Map((await listInstances(options.instances)).map((i) => [i.id, i.name])); return { items: page.items.map( (item): FleetOutboxItem => ({ ...item, instanceName: names.get(item.instanceId) ?? item.instanceId, }), ), total: page.total, summary: options.outbox.summary(), }; }), ); app.post( '/api/v1/fleet/messages/outbox/flush', wrapMessageAction(async () => { if (!options.outbox) return { attempted: 0, delivered: 0, deferred: 0, failed: 0, remaining: 0 }; return await options.outbox.flush(); }), ); app.post( '/api/v1/fleet/messages/outbox/:queueId/cancel', wrapMessageAction(async (request) => { if (!options.outbox) throw new MessageServiceError('NOT_FOUND'); const params = record(request.params); return options.outbox.cancel(outboxQueueId(params?.queueId)); }), ); app.post( '/api/v1/fleet/messages/outbox/:queueId/retry', wrapMessageAction(async (request) => { if (!options.outbox) throw new MessageServiceError('NOT_FOUND'); const params = record(request.params); return options.outbox.retry(outboxQueueId(params?.queueId)); }), ); app.delete( '/api/v1/fleet/messages/outbox/:queueId', wrapMessageAction(async (request) => { if (!options.outbox) throw new MessageServiceError('NOT_FOUND'); const params = record(request.params); const queueId = outboxQueueId(params?.queueId); options.outbox.remove(queueId); return { removed: true, queueId }; }), ); }