feat: add Telegram private messaging channel

Connect a user-created Telegram bot to channel projects with encrypted credentials, numeric sender allowlists, unified execution and text replies. Use Electron networking for system proxy support, with polling recovery, cancellable delivery retries and live status updates.

Add schema 61 channel support, lifecycle regression tests, Electron integration probes and feature documentation. Require the performance principles in AGENTS.md and record three alternating baseline/current Main benchmark rounds.

Validation: typecheck, lint and diff checks passed. Full suite: 6087 passed, 87 skipped, one intermittent inline-help Electron failure that passed unchanged in isolation. Real Bot plus model acceptance remains pending; the local TLS proxy probe is opt-in.

Release note: 新增 Telegram 私聊通道,可配置 Bot Token 和用户白名单,通过通道项目处理文字请求并回复结果。支持系统代理、断线重连和停用取消;真实 Bot 与模型的完整联调仍待完成。
This commit is contained in:
mesalogo
2026-10-05 02:42:28 +08:00
parent 78afec7775
commit cf598121b1
56 changed files with 4144 additions and 194 deletions
@@ -2841,6 +2841,14 @@ describe('AssistantDatabase', () => {
expect.objectContaining({
kind: 'channel',
channel: 'dingtalk'
}),
expect.objectContaining({
name: 'Telegram',
kind: 'channel',
channel: 'telegram',
rootPath: 'C:\\Users\\test',
executionSpace: { kind: 'local', rootPath: 'C:\\Users\\test' },
runtimeSelection: { provider: 'model' }
})
])
expect(second.map((project) => project.id)).toEqual(
@@ -2891,6 +2899,101 @@ describe('AssistantDatabase', () => {
database.close()
})
it('relaxes schema 60 channel checks without rewriting or losing existing records', async () => {
const directory = await mkdtemp(join(tmpdir(), 'goodbuddy-telegram-migration-'))
temporaryDirectories.push(directory)
const path = join(directory, 'assistant.sqlite')
const original = new AssistantDatabase(path)
original.initialize('C:\\Workspace')
const projects = original.ensureChannelProjects('C:\\Channels')
for (const project of projects.filter(project => project.channel !== 'telegram')) {
const conversation = original.getOrCreateRemoteConversation({
projectId: project.id, channel: project.channel!, accountId: 'legacy-account',
externalConversationId: 'legacy-chat', conversationType: 'direct',
title: 'Legacy conversation', accountDisplay: 'Legacy account',
runtimeSelection: { provider: 'model' }
})
original.appendRemoteConversationMessage({ conversationId: conversation.id, role: 'user', content: 'Preserve this message' })
}
original.close()
const legacy = new DatabaseSync(path)
legacy.enableDefensive(false)
legacy.exec(`PRAGMA foreign_keys = ON; DELETE FROM projects WHERE channel = 'telegram';
PRAGMA writable_schema = ON;
UPDATE sqlite_schema SET sql = replace(sql,
'channel IN (''weixin'', ''wecom'', ''dingtalk'', ''telegram'')',
'channel IN (''weixin'', ''wecom'', ''dingtalk'')')
WHERE type = 'table' AND name IN ('projects', 'conversations');
PRAGMA writable_schema = RESET; PRAGMA user_version = 60;`)
const tables = ['projects', 'project_execution_spaces', 'conversations', 'messages']
const rows = tables.map(table => legacy.prepare(`SELECT rowid, * FROM ${table} ORDER BY rowid`).all())
const schema = legacy.prepare('SELECT type, name, tbl_name, rootpage, sql FROM sqlite_schema ORDER BY name').all()
expect(() => legacy.prepare("UPDATE projects SET channel = 'telegram' WHERE channel = 'weixin'").run()).toThrow(/CHECK/)
legacy.close()
const migrated = new AssistantDatabase(path)
const progress = vi.fn()
migrated.upgradeStorage(progress, () => false, { migrateNotes: false, reclaimSpace: false })
expect(progress).not.toHaveBeenCalled()
const check = new DatabaseSync(path)
expect(check.prepare('PRAGMA user_version').get()!.user_version).toBe(61)
expect(tables.map(table => check.prepare(`SELECT rowid, * FROM ${table} ORDER BY rowid`).all())).toEqual(rows)
expect(check.prepare('SELECT type, name, tbl_name, rootpage, sql FROM sqlite_schema ORDER BY name').all()).toEqual(
schema.map(row => ['projects', 'conversations'].includes(row.name as string) && row.type === 'table'
? { ...row, sql: (row.sql as string).replace("channel IN ('weixin', 'wecom', 'dingtalk')", "channel IN ('weixin', 'wecom', 'dingtalk', 'telegram')") }
: row)
)
expect(check.prepare('PRAGMA integrity_check').get()!.integrity_check).toBe('ok')
expect(check.prepare('PRAGMA foreign_key_check').all()).toEqual([])
expect(() => check.prepare("UPDATE projects SET channel = 'invalid' WHERE channel = 'weixin'").run()).toThrow(/CHECK/)
expect(() => check.prepare("UPDATE conversations SET channel = 'invalid' WHERE channel = 'weixin'").run()).toThrow(/CHECK/)
check.close()
migrated.initialize('C:\\Workspace')
const telegram = migrated.ensureChannelProjects('C:\\Telegram').find(project => project.channel === 'telegram')!
const input = {
projectId: telegram.id, channel: 'telegram' as const, accountId: 'bot',
externalConversationId: '-100123', conversationType: 'group' as const,
title: 'Telegram chat', accountDisplay: 'Telegram', runtimeSelection: { provider: 'model' as const }
}
const conversation = migrated.getOrCreateRemoteConversation(input)
migrated.close()
const reopened = new AssistantDatabase(path)
reopened.initialize('C:\\Workspace')
expect(reopened.ensureChannelProjects('C:\\Ignored').find(project => project.channel === 'telegram')).toEqual(telegram)
expect(reopened.getOrCreateRemoteConversation(input).id).toBe(conversation.id)
expect(reopened.getConversation(conversation.id).remote?.channel).toBe('telegram')
reopened.close()
})
it('rolls back the channel CHECK migration if a schema definition is unexpected', async () => {
const directory = await mkdtemp(join(tmpdir(), 'goodbuddy-telegram-rollback-'))
temporaryDirectories.push(directory)
const path = join(directory, 'assistant.sqlite')
const database = new AssistantDatabase(path)
database.initialize('C:\\Workspace')
database.close()
const legacy = new DatabaseSync(path)
legacy.enableDefensive(false)
legacy.exec(`PRAGMA writable_schema = ON;
UPDATE sqlite_schema SET sql = replace(sql,
'channel IN (''weixin'', ''wecom'', ''dingtalk'', ''telegram'')',
'channel IN (''weixin'', ''wecom'', ''dingtalk'')') WHERE name = 'projects';
UPDATE sqlite_schema SET sql = replace(sql,
'channel IN (''weixin'', ''wecom'', ''dingtalk'', ''telegram'')',
'channel IN (''weixin'')') WHERE name = 'conversations';
PRAGMA writable_schema = RESET; PRAGMA user_version = 60;`)
const before = legacy.prepare('SELECT * FROM sqlite_schema ORDER BY name').all()
legacy.close()
expect(() => database.upgradeStorage(() => undefined, () => false, { migrateNotes: false, reclaimSpace: false }))
.toThrow('Unrecognized channel constraint: conversations')
const check = new DatabaseSync(path)
expect(check.prepare('PRAGMA user_version').get()!.user_version).toBe(60)
expect(check.prepare('SELECT * FROM sqlite_schema ORDER BY name').all()).toEqual(before)
expect(check.prepare('PRAGMA integrity_check').get()!.integrity_check).toBe('ok')
check.close()
})
it('persists one protected remote conversation per channel identity', async () => {
const database = await createDatabase()
const project = database.ensureChannelProjects(
@@ -3099,6 +3202,31 @@ describe('AssistantDatabase', () => {
database.close()
})
it('persists an immediate terminal result without attachments or repeated attempt counts', async () => {
const directory = await mkdtemp(join(tmpdir(), 'goodbuddy-channel-terminal-'))
temporaryDirectories.push(directory)
const path = join(directory, 'assistant.sqlite')
const database = new AssistantDatabase(path)
database.initialize('C:\\Workspace')
const entry = database.enqueueChannelResult({
channel: 'telegram', eventId: 'permanent-failure', conversationId: 'chat',
recipientId: 'user', status: 'completed', output: 'Keep result text',
attachments: [{ name: 'result.txt', mimeType: 'text/plain', size: 1, kind: 'file', dataBase64: 'eA==' }]
})
database.markChannelResult(entry.id, 'terminal')
database.close()
const reopened = new AssistantDatabase(path)
reopened.initialize('C:\\Workspace')
reopened.markChannelResult(entry.id, 'terminal')
reopened.markChannelResult(entry.id, 'failed')
expect(reopened.listUndeliveredChannelResults('telegram')).toEqual([
expect.objectContaining({ id: entry.id, state: 'terminal', attempts: 1,
message: expect.objectContaining({ output: 'Keep result text' }) })
])
expect(reopened.listUndeliveredChannelResults('telegram')[0]!.message).not.toHaveProperty('attachments')
reopened.close()
})
it('migrates exhausted legacy outbox failures to terminal state', async () => {
const directory = await mkdtemp(
join(tmpdir(), 'goodbuddy-channel-terminal-migration-')
+44 -9
View File
@@ -136,7 +136,7 @@ import {
SUBAGENT_PROGRESS_STORAGE_SCHEMA_VERSION
} from './subagent-progress-storage'
export const ASSISTANT_DATABASE_SCHEMA_VERSION = 60
export const ASSISTANT_DATABASE_SCHEMA_VERSION = 61
function withoutHistoricalWorkMode(value: unknown): unknown {
if (!value || typeof value !== 'object' || Array.isArray(value)) return value
@@ -2623,6 +2623,11 @@ export class AssistantDatabase {
channel: 'dingtalk',
name: '钉钉',
description: '钉钉远程消息与受控任务'
},
{
channel: 'telegram',
name: 'Telegram',
description: 'Telegram 远程消息与受控任务'
}
]
const find = database.prepare(
@@ -4084,7 +4089,7 @@ export class AssistantDatabase {
markChannelResult(
id: string,
state: 'delivered' | 'failed'
state: 'delivered' | 'failed' | 'terminal'
): void {
this.requireDatabase()
.prepare(
@@ -4096,11 +4101,11 @@ export class AssistantDatabase {
END,
attempts = attempts + 1,
message_json = CASE
WHEN ? = 'delivered' OR attempts + 1 >= 5
WHEN ? IN ('delivered', 'terminal') OR attempts + 1 >= 5
THEN json_remove(message_json, '$.attachments')
ELSE message_json
END
WHERE id = ?`
WHERE id = ? AND state NOT IN ('delivered', 'terminal')`
)
.run(state, state, state, id)
}
@@ -4124,10 +4129,10 @@ export class AssistantDatabase {
`WITH pending AS (
SELECT id, message_json, state, attempts, created_at,
ROW_NUMBER() OVER (
ORDER BY attempts ASC, created_at ASC
ORDER BY (state = 'terminal') ASC, attempts ASC, created_at ASC, id ASC
) AS position,
SUM(LENGTH(CAST(message_json AS BLOB))) OVER (
ORDER BY attempts ASC, created_at ASC
ORDER BY (state = 'terminal') ASC, attempts ASC, created_at ASC, id ASC
) AS cumulative_bytes
FROM channel_outbox
WHERE state != 'delivered'
@@ -4136,7 +4141,7 @@ export class AssistantDatabase {
SELECT id, message_json, state, attempts, created_at
FROM pending
WHERE position = 1 OR cumulative_bytes <= ?
ORDER BY attempts ASC, created_at ASC
ORDER BY (state = 'terminal') ASC, attempts ASC, created_at ASC, id ASC
LIMIT ?`
)
.all(
@@ -10663,7 +10668,7 @@ export class AssistantDatabase {
ADD COLUMN channel TEXT
CHECK(
channel IS NULL OR
channel IN ('weixin', 'wecom', 'dingtalk')
channel IN ('weixin', 'wecom', 'dingtalk', 'telegram')
);
`)
}
@@ -10694,7 +10699,7 @@ export class AssistantDatabase {
ALTER TABLE conversations ADD COLUMN channel TEXT
CHECK(
channel IS NULL OR
channel IN ('weixin', 'wecom', 'dingtalk')
channel IN ('weixin', 'wecom', 'dingtalk', 'telegram')
);
`)
}
@@ -12299,6 +12304,36 @@ export class AssistantDatabase {
`)
} catch (error) { database.exec('ROLLBACK'); throw error }
}
if (version.user_version < 61) {
database.exec('BEGIN IMMEDIATE')
try {
// SQLite's metadata-only CHECK relaxation preserves table rows, rowids,
// indexes and references without rebuilding potentially large conversations.
const previous = "channel IN ('weixin', 'wecom', 'dingtalk')"
const expanded = "channel IN ('weixin', 'wecom', 'dingtalk', 'telegram')"
const schemaVersion = database.prepare('PRAGMA schema_version').get() as { schema_version: number }
database.enableDefensive(false)
for (const table of ['projects', 'conversations']) {
const definition = database.prepare("SELECT sql FROM sqlite_schema WHERE type = 'table' AND name = ?").get(table) as { sql: string }
if (definition.sql.includes(expanded)) continue
if (!definition.sql.includes(previous)) throw new Error(`Unrecognized channel constraint: ${table}`)
database.exec('PRAGMA writable_schema = ON')
database.prepare("UPDATE sqlite_schema SET sql = ? WHERE type = 'table' AND name = ?")
.run(definition.sql.replace(previous, expanded), table)
}
database.exec(`PRAGMA schema_version = ${schemaVersion.schema_version + 1}; PRAGMA writable_schema = RESET;`)
// Reparse with writable_schema disabled so invalid definitions roll back.
database.prepare('SELECT channel FROM projects LIMIT 0').all()
database.prepare('SELECT channel FROM conversations LIMIT 0').all()
database.exec('PRAGMA user_version = 61; COMMIT;')
} catch (error) {
database.exec('ROLLBACK')
throw error
} finally {
database.exec('PRAGMA writable_schema = RESET')
database.enableDefensive(true)
}
}
}
supervisionExperiences(): SupervisionExperienceStore {
@@ -110,7 +110,7 @@ describe('subagent progress storage', () => {
for (const [table, column] of [['projects', 'default_work_mode'], ['conversations', 'work_mode'], ['tasks', 'work_mode']]) {
expect(check.prepare(`PRAGMA table_info(${table})`).all().map(row => row.name)).not.toContain(column)
}
expect(check.prepare('PRAGMA user_version').get()!.user_version).toBe(60)
expect(check.prepare('PRAGMA user_version').get()!.user_version).toBe(ASSISTANT_DATABASE_SCHEMA_VERSION)
const readJson = (table: string, column: string, id: string) =>
JSON.parse(check.prepare(`SELECT ${column} AS value FROM ${table} WHERE id = ?`).get(id)!.value as string)
expect(readJson('conversations', 'context_state_json', header.id))
@@ -164,7 +164,7 @@ describe('subagent progress storage', () => {
sql.prepare('UPDATE task_events SET payload_json = ? WHERE id = ?').run('{"workMode":"ask"}', damagedId)
upgradeAssistantStorage(path, () => undefined)
expect(sql.prepare('SELECT payload_json FROM task_events WHERE id = ?').get(damagedId)!.payload_json).toBe('{}')
expect(sql.prepare('PRAGMA user_version').get()!.user_version).toBe(60)
expect(sql.prepare('PRAGMA user_version').get()!.user_version).toBe(ASSISTANT_DATABASE_SCHEMA_VERSION)
expect(sql.prepare('PRAGMA integrity_check').get()!.integrity_check).toBe('ok')
} finally { sql.close() }
})
+21
View File
@@ -19,6 +19,14 @@ export interface ChannelDriver {
stop(): void | Promise<void>
}
export function isPermanentChannelError(error: unknown): boolean {
return (
error instanceof Error &&
'permanent' in error &&
error.permanent === true
)
}
export interface DedupStore {
claim(
channel: string,
@@ -83,6 +91,7 @@ export interface Outbox {
enqueue(message: ChannelResultMessage): OutboxEntry | Promise<OutboxEntry>
markDelivered(id: string): void | Promise<void>
markFailed(id: string): void | Promise<void>
markTerminal?(id: string): void | Promise<void>
listUndelivered(
channel?: string,
limit?: number
@@ -134,6 +143,16 @@ export class MemoryOutbox implements Outbox {
}
}
markTerminal(id: string): void {
const entry = this.entries.get(id)
if (!entry) {
return
}
entry.state = 'terminal'
entry.attempts += 1
entry.message = this.withoutAttachments(entry.message)
}
listUndelivered(
channel?: string,
limit = this.maximumEntries
@@ -146,6 +165,8 @@ export class MemoryOutbox implements Outbox {
)
.sort(
(left, right) =>
Number(left.state === 'terminal') -
Number(right.state === 'terminal') ||
left.attempts - right.attempts ||
left.createdAt - right.createdAt
)
+370 -2
View File
@@ -1,8 +1,9 @@
import { mkdtemp, rm } from 'node:fs/promises'
import { mkdtemp, readFile, rm } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { afterEach, describe, expect, it, vi } from 'vitest'
import type { ChannelDriver } from './channel-driver'
import type { ChannelDriver, ChannelExecutor, ChannelInboundHandler } from './channel-driver'
import type { ChannelRuntimeStatus, ChannelRuntimeStatusChange } from '../../shared/channel-settings-contracts'
import {
ChannelManager,
type ManagedChannelService
@@ -15,7 +16,11 @@ import {
const roots: string[] = []
const electronFetch = vi.hoisted(() => vi.fn<typeof globalThis.fetch>())
vi.mock('electron', () => ({ net: { fetch: electronFetch } }))
afterEach(async () => {
electronFetch.mockReset()
await Promise.all(
roots.splice(0).map((root) => rm(root, { recursive: true, force: true }))
)
@@ -102,6 +107,369 @@ function managerHarness(
}
describe('ChannelManager', () => {
it('emits deduplicated status deltas for async transitions and ignores late callbacks', async () => {
const settingsStore = await store()
const input = { enabled: true, allowedSenderIds: [], allowGroupMessages: false as const }
await settingsStore.apply({ telegram: {
...input, secret: { action: 'replace', value: 'status-event-test-secret' }
} })
const changes: ChannelRuntimeStatusChange[] = []
const callbacks: Array<(status: ChannelRuntimeStatus) => void> = []
const manager = new ChannelManager(settingsStore, executor, {
onStatusChanged: (change) => { changes.push(change) },
createDriver: (settings, onStatus) => {
if (onStatus) callbacks.push(onStatus)
return {
...inertDriver(settings.channel),
testConnection: async () => ({ botId: '123' })
}
},
createService: () => {
const onStatus = callbacks.at(-1)!
return {
start: async () => { onStatus({ state: 'starting' }) },
stop: async () => { onStatus({ state: 'stopped' }) }
}
}
})
await manager.initialize()
expect(changes).toEqual([
{ channel: 'weixin', status: { state: 'disabled' } },
{ channel: 'wecom', status: { state: 'disabled' } },
{ channel: 'dingtalk', status: { state: 'disabled' } },
{ channel: 'telegram', status: { state: 'starting' } }
])
changes.length = 0
callbacks[0]!({ state: 'running' })
callbacks[0]!({ state: 'running' })
callbacks[0]!({ state: 'error', lastError: 'status-event-test-secret disconnected' })
callbacks[0]!({ state: 'error', lastError: 'status-event-test-secret disconnected' })
callbacks[0]!({ state: 'error', lastError: 'polling conflict' })
callbacks[0]!({ state: 'running' })
expect(changes.map((change) => change.status.state)).toEqual(['running', 'error', 'error', 'running'])
expect(changes.every((change) => change.channel === 'telegram')).toBe(true)
expect(JSON.stringify(changes)).not.toContain('status-event-test-secret')
expect(changes.at(-1)).toEqual({ channel: 'telegram', status: { state: 'running' } })
changes.length = 0
await manager.testConnection('telegram')
expect(changes).toEqual([])
await manager.reload('telegram')
callbacks[0]!({ state: 'error', lastError: 'retired' })
callbacks[1]!({ state: 'running' })
expect(changes).toEqual([
{ channel: 'telegram', status: { state: 'starting' } },
{ channel: 'telegram', status: { state: 'running' } }
])
changes.length = 0
await expect(manager.apply({ telegram: { ...input, secret: { action: 'clear' } } })).rejects.toThrow('Bot Token')
expect((await settingsStore.resolve('telegram')).secret).toBe('status-event-test-secret')
expect(changes).toEqual([])
await manager.apply({ telegram: { ...input, enabled: false, secret: { action: 'clear' } } })
callbacks[1]!({ state: 'running' })
await manager.reload('telegram')
await manager.stopAll()
expect(changes).toEqual([{ channel: 'telegram', status: { state: 'disabled' } }])
expect((await settingsStore.resolve('telegram')).secret).toBeUndefined()
})
it('leaves Telegram connection status to the driver during startup and delivery', async () => {
const settingsStore = await store()
await settingsStore.apply({ telegram: {
enabled: true, secret: { action: 'replace', value: 'status-test-secret' },
allowedSenderIds: [], allowGroupMessages: false
} })
let status!: (status: ChannelRuntimeStatus) => void
let deliverySuccess!: () => void
let deliveryFailure!: (error: Error) => void
const manager = new ChannelManager(settingsStore, executor, {
createDriver: (settings, onStatus) => {
status = onStatus!
return inertDriver(settings.channel)
},
createService: (_driver, _executor, options) => {
deliverySuccess = () => options.onDeliverySuccess?.()
deliveryFailure = (error) => options.onDeliveryFailure?.(error)
return { start: async () => undefined, stop: async () => undefined }
}
})
expect((await manager.initialize()).telegram.status).toEqual({ state: 'starting' })
deliverySuccess()
expect((await manager.snapshot()).telegram.status.state).toBe('starting')
status({ state: 'error', lastError: 'Telegram polling conflict (409)' })
deliverySuccess()
deliveryFailure(new Error('send failed'))
expect((await manager.snapshot()).telegram.status).toEqual({
state: 'error', lastError: 'Telegram polling conflict (409)'
})
status({ state: 'running' })
deliveryFailure(new Error('send failed'))
expect((await manager.snapshot()).telegram.status).toEqual({ state: 'running' })
await manager.stopAll()
})
it('updates Telegram whitelists without cancelling accepted work or restarting the service', async () => {
const settingsStore = await store()
const input = { enabled: true, allowedSenderIds: ['123'], allowGroupMessages: false as const }
await settingsStore.apply({ telegram: { ...input, secret: { action: 'replace', value: 'active-test-secret' } } })
let inbound!: ChannelInboundHandler
let activeSignal!: AbortSignal
let finish!: () => void
const work = new Promise<void>((resolve) => { finish = resolve })
const execute = vi.fn<ChannelExecutor>(async (_message, signal) => {
activeSignal = signal
await work
return { status: 'completed', output: 'finished' }
})
const updateAllowedSenderIds = vi.fn()
const stop = vi.fn(async () => undefined)
const send = vi.fn(async () => undefined)
const createDriver = vi.fn((settings: ResolvedChannelSettings) => ({
...inertDriver(settings.channel),
start: async (handler: ChannelInboundHandler) => { inbound = handler },
updateAllowedSenderIds, stop, send
}))
const manager = new ChannelManager(settingsStore, execute, { createDriver })
await manager.initialize()
const message = {
channel: 'telegram', accountId: '456', eventId: '1', senderId: '123',
conversationId: '456:123', conversationType: 'direct', text: 'hello'
}
await inbound(message, () => undefined)
await vi.waitFor(() => expect(execute).toHaveBeenCalledOnce())
await manager.apply({ telegram: { ...input, allowedSenderIds: [], secret: { action: 'keep' } } })
expect(updateAllowedSenderIds).toHaveBeenLastCalledWith([])
expect(createDriver).toHaveBeenCalledOnce()
expect(stop).not.toHaveBeenCalled()
expect(activeSignal.aborted).toBe(false)
await inbound({ ...message, eventId: '2' }, () => undefined)
finish()
await vi.waitFor(() => expect(send).toHaveBeenCalled())
expect(execute).toHaveBeenCalledOnce()
await manager.apply({ telegram: { ...input, allowedSenderIds: ['789'], secret: { action: 'keep' } } })
await inbound({ ...message, eventId: '3', senderId: '789', conversationId: '456:789' }, () => undefined)
await vi.waitFor(() => expect(execute).toHaveBeenCalledTimes(2))
expect(createDriver).toHaveBeenCalledOnce()
await manager.stopAll()
})
it.each(['keep', 'replace'] as const)('restarts a conflicted Telegram poller when saving the same token with %s', async (action) => {
const settingsStore = await store()
const input = { enabled: true, allowedSenderIds: ['123'], allowGroupMessages: false as const }
const secret = 'recovery-test-placeholder'
await settingsStore.apply({ telegram: { ...input, secret: { action: 'replace', value: secret } } })
let conflict = true
let polls = 0
electronFetch.mockImplementation(async (input, init) => {
const method = String(input).split('/').at(-1)
if (method === 'getMe') return Response.json({ ok: true, result: { id: 456, is_bot: true } })
if (method === 'getWebhookInfo') return Response.json({ ok: true, result: { url: '' } })
if (method === 'getUpdates') {
polls++
if (conflict) return Response.json({ ok: false }, { status: 409 })
return new Promise<Response>((_resolve, reject) => {
const signal = init!.signal!
if (signal.aborted) reject(signal.reason)
else signal.addEventListener('abort', () => reject(signal.reason), { once: true })
})
}
throw new Error(`Unexpected method: ${method}`)
})
const manager = new ChannelManager(settingsStore, executor)
try {
await manager.initialize()
await vi.waitFor(async () => expect((await manager.snapshot()).telegram.status.state).toBe('error'))
conflict = false
expect((await manager.testConnection('telegram')).ok).toBe(true)
await manager.apply({ telegram: { ...input, secret: action === 'keep' ? { action } : { action, value: secret } } })
await vi.waitFor(async () => {
expect(polls).toBe(2)
expect((await manager.snapshot()).telegram.status.state).toBe('running')
})
} finally {
await manager.stopAll()
}
})
it('validates replacement tokens before saving or stopping the active Telegram service', async () => {
const root = await mkdtemp(join(tmpdir(), 'goodbuddy-manager-'))
roots.push(root)
const path = join(root, 'settings.json')
const settingsStore = new ChannelSettingsStore(path, cipher(), {})
const input = { enabled: true, allowedSenderIds: ['123'], allowGroupMessages: false as const }
await settingsStore.apply({ telegram: { ...input, secret: { action: 'replace', value: 'working-test-secret' } } })
const original = await readFile(path, 'utf8')
const events: string[] = []
const probes: Array<{ start: ReturnType<typeof vi.fn>; stop: ReturnType<typeof vi.fn> }> = []
let activeStatus!: (status: ChannelRuntimeStatus) => void
const updateAllowedSenderIds = vi.fn()
const manager = new ChannelManager(settingsStore, executor, {
createDriver: (settings, onStatus) => {
const secret = settings.channel === 'telegram' ? settings.secret : ''
if (onStatus) {
activeStatus = onStatus
return { ...inertDriver(settings.channel), updateAllowedSenderIds }
}
const driver = {
...inertDriver(settings.channel), start: vi.fn(), stop: vi.fn(() => { events.push('probe-stop') }),
testConnection: vi.fn(async () => {
events.push('probe')
expect(await readFile(path, 'utf8')).toBe(original)
expect((await manager.snapshot()).telegram.status.state).toBe('running')
if (secret === 'invalid-test-secret') throw new Error('invalid-test-secret invalid token')
return { botId: '456' }
})
}
probes.push(driver)
return driver
},
createService: () => ({
start: async () => { events.push('service-start'); activeStatus({ state: 'running' }) },
stop: async () => { events.push('service-stop') },
updateAllowedSenderIds
})
})
await manager.initialize()
events.length = 0
await expect(manager.apply({ telegram: {
...input, secret: { action: 'replace', value: 'invalid-test-secret' }
} })).rejects.not.toThrow('invalid-test-secret')
expect(events).toEqual(['probe', 'probe-stop'])
expect(await readFile(path, 'utf8')).toBe(original)
expect((await settingsStore.resolve('telegram')).secret).toBe('working-test-secret')
expect((await manager.snapshot()).telegram.status.state).toBe('running')
events.length = 0
await manager.apply({ telegram: { ...input, secret: { action: 'replace', value: 'valid-test-secret' } } })
expect(events).toEqual(['probe', 'probe-stop', 'service-stop', 'service-start'])
expect((await settingsStore.resolve('telegram')).secret).toBe('valid-test-secret')
for (const probe of probes) {
expect(probe.start).not.toHaveBeenCalled()
expect(probe.stop).toHaveBeenCalledOnce()
}
expect(updateAllowedSenderIds).not.toHaveBeenCalled()
await manager.stopAll()
})
it('uses Electron fetch in the default Telegram factory and only probes identity and webhook', async () => {
electronFetch.mockImplementation(async (input) => {
const method = String(input).split('/').at(-1)
if (method === 'getMe') {
return new Response(JSON.stringify({ ok: true, result: { id: 123, is_bot: true, username: 'example_bot' } }))
}
if (method === 'getWebhookInfo') {
return new Response(JSON.stringify({ ok: true, result: { url: '' } }))
}
throw new Error(`Unexpected Telegram method: ${method}`)
})
const manager = new ChannelManager(await store(), executor)
expect(await manager.testConnection('telegram', {
enabled: false, secret: { action: 'replace', value: 'test-only-secret' },
allowedSenderIds: [], allowGroupMessages: false
})).toEqual({ channel: 'telegram', ok: true, botUsername: 'example_bot' })
expect(electronFetch.mock.calls.map(([input]) => String(input).split('/').at(-1)))
.toEqual(['getMe', 'getWebhookInfo'])
expect((await manager.snapshot()).telegram.status.state).toBe('disabled')
})
it('tests Telegram identity without creating a second poller or changing active settings', async () => {
const settingsStore = await store()
const starts = vi.fn(async () => undefined)
const stops = vi.fn(async () => undefined)
const probe = vi.fn(async () => ({ botId: '123', botUsername: 'example_bot' }))
const createService = vi.fn(() => ({ start: starts, stop: stops }))
const testDrivers: ChannelDriver[] = []
const manager = new ChannelManager(settingsStore, executor, {
createDriver: (settings, onStatus) => {
onStatus?.({ state: 'running' })
const driver = { ...inertDriver(settings.channel), start: vi.fn(), stop: vi.fn(), testConnection: probe }
testDrivers.push(driver)
return driver
},
createService
})
const input = { enabled: true, allowedSenderIds: [], allowGroupMessages: false as const }
await settingsStore.apply({ telegram: { ...input, secret: { action: 'replace', value: 'stored-test-secret' } } })
await manager.initialize()
expect((await manager.snapshot()).telegram.status.state).toBe('running')
expect(await manager.testConnection('telegram', {
...input, secret: { action: 'replace', value: 'temporary-test-secret' }
})).toEqual({ channel: 'telegram', ok: true, botUsername: 'example_bot' })
expect(await manager.test('telegram')).toEqual({ channel: 'telegram', ok: true, botUsername: 'example_bot' })
expect(probe).toHaveBeenCalledTimes(2)
expect(createService).toHaveBeenCalledOnce()
expect(starts).toHaveBeenCalledOnce()
expect(stops).not.toHaveBeenCalled()
for (const driver of testDrivers.slice(1)) {
expect(driver.start).not.toHaveBeenCalled()
expect(driver.stop).toHaveBeenCalledOnce()
}
expect((await settingsStore.resolve('telegram')).secret).toBe('stored-test-secret')
expect((await manager.snapshot()).telegram.status.state).toBe('running')
probe.mockRejectedValueOnce(new Error('temporary-test-secret failed'))
const failure = await manager.test('telegram', { ...input, secret: { action: 'replace', value: 'temporary-test-secret' } })
expect(failure.ok).toBe(false)
expect(failure.error).not.toContain('temporary-test-secret')
await manager.stopAll()
})
it('tracks Telegram runtime statuses and ignores retired driver and service callbacks', async () => {
const settingsStore = await store()
const callbacks: Array<(status: ChannelRuntimeStatus) => void> = []
const deliveryCallbacks: Array<() => void> = []
const lifecycle: string[] = []
const manager = new ChannelManager(settingsStore, executor, {
createDriver: (settings, onStatus) => {
callbacks.push(onStatus!)
return inertDriver(settings.channel)
},
createService: (_driver, _executor, options) => {
const index = callbacks.length - 1
deliveryCallbacks.push(() => options.onDeliverySuccess?.())
return {
start: async () => {
lifecycle.push(`start-${index}`)
callbacks[index]?.({ state: 'running' })
},
stop: async () => {
lifecycle.push(`stop-${index}`)
callbacks[index]?.({ state: 'error', lastError: 'retired' })
}
}
}
})
const input = { enabled: true, allowedSenderIds: [], allowGroupMessages: false as const }
await settingsStore.apply({ telegram: { ...input, secret: { action: 'replace', value: 'status-test-secret' } } })
await manager.initialize()
callbacks[0]?.({ state: 'error', lastError: 'status-test-secret network failure' })
expect((await manager.snapshot()).telegram.status).toMatchObject({ state: 'error' })
expect((await manager.snapshot()).telegram.status.lastError).not.toContain('status-test-secret')
callbacks[0]?.({ state: 'running' })
expect((await manager.snapshot()).telegram.status).toEqual({ state: 'running' })
await manager.apply({ telegram: { ...input, secret: { action: 'keep' }, allowedSenderIds: ['123'] } })
expect(lifecycle).toEqual(['start-0', 'stop-0', 'start-1'])
callbacks[0]?.({ state: 'error', lastError: 'late failure' })
expect((await manager.snapshot()).telegram.status.state).toBe('running')
await manager.apply({ telegram: { ...input, enabled: false, secret: { action: 'keep' } } })
callbacks[1]?.({ state: 'running' })
deliveryCallbacks[1]?.()
expect((await manager.snapshot()).telegram.status.state).toBe('disabled')
await manager.apply({ telegram: { ...input, secret: { action: 'keep' } } })
await manager.stopAll()
callbacks[2]?.({ state: 'running' })
expect((await manager.snapshot()).telegram.status.state).toBe('stopped')
})
it('rejects temporary Telegram environment overrides', async () => {
const root = await mkdtemp(join(tmpdir(), 'goodbuddy-manager-'))
roots.push(root)
const settingsStore = new ChannelSettingsStore(join(root, 'settings.json'), cipher(), {
GOODBUDDY_TELEGRAM_BOT_TOKEN: 'environment-test-secret'
})
const createDriver = vi.fn(() => inertDriver('telegram'))
const manager = new ChannelManager(settingsStore, executor, { createDriver })
expect((await manager.testConnection('telegram', {
enabled: true, secret: { action: 'keep' }, allowedSenderIds: [], allowGroupMessages: false
})).ok).toBe(false)
expect(createDriver).not.toHaveBeenCalled()
})
it('applies settings and dynamically starts, replaces, and disables services', async () => {
const settingsStore = await store()
const { manager, services } = managerHarness(settingsStore)
+156 -26
View File
@@ -1,15 +1,19 @@
import {
CHANNEL_SETTINGS_LIMITS,
channelConnectionTestResultSchema,
channelSettingsApplySchema,
dingTalkChannelSettingsInputSchema,
telegramChannelSettingsInputSchema,
weComChannelSettingsInputSchema,
type ChannelConnectionTestResult,
type ChannelRuntimeStatus,
type ChannelRuntimeStatusChange,
type ChannelSettingsApply,
type ChannelSettingsSnapshot,
type CredentialChannel,
type DingTalkChannelSettingsInput,
type ManagedChannel,
type TelegramChannelSettingsInput,
type WeComChannelSettingsInput
} from '../../shared/channel-settings-contracts'
import type {
@@ -33,11 +37,17 @@ import type { WechatSidecarLauncher } from './wechat-sidecar-client'
export type ManagedChannelService = Pick<
ChannelService,
'start' | 'stop'
>
> & { updateAllowedSenderIds?: (ids: readonly string[]) => void }
export type ChannelDriverFactory = (
settings: ResolvedChannelSettings
) => ChannelDriver | Promise<ChannelDriver>
settings: ResolvedChannelSettings,
onStatus?: (status: ChannelRuntimeStatus) => void
) => TestableChannelDriver | Promise<TestableChannelDriver>
type TestableChannelDriver = ChannelDriver & {
testConnection?: () => Promise<{ botId: string; botUsername?: string }>
updateAllowedSenderIds?: (ids: readonly string[]) => void
}
export type ChannelServiceFactory = (
driver: ChannelDriver,
@@ -53,6 +63,7 @@ export type ChannelServiceFactory = (
) => ManagedChannelService | Promise<ManagedChannelService>
export type ChannelManagerOptions = {
onStatusChanged?: (change: ChannelRuntimeStatusChange) => void
createDriver?: ChannelDriverFactory
createService?: ChannelServiceFactory
launchWechatSidecar?: WechatSidecarLauncher
@@ -61,6 +72,10 @@ export type ChannelManagerOptions = {
}
type TestSettingsInput =
| {
channel: 'telegram'
settings?: TelegramChannelSettingsInput
}
| {
channel: 'wecom'
settings?: WeComChannelSettingsInput
@@ -70,10 +85,11 @@ type TestSettingsInput =
settings?: DingTalkChannelSettingsInput
}
function defaultDriverFactory(
async function defaultDriverFactory(
settings: ResolvedChannelSettings,
launchWechatSidecar?: WechatSidecarLauncher
): ChannelDriver {
launchWechatSidecar?: WechatSidecarLauncher,
onStatus?: (status: ChannelRuntimeStatus) => void
): Promise<TestableChannelDriver> {
if (settings.channel === 'weixin') {
if (!launchWechatSidecar) {
throw new Error('微信 Sidecar 启动器不可用')
@@ -83,6 +99,18 @@ function defaultDriverFactory(
if (settings.secret === undefined) {
throw new Error('通道 Secret 尚未配置')
}
if (settings.channel === 'telegram') {
const [{ TelegramChannelDriver }, { net }] = await Promise.all([
import('./telegram-channel-driver'),
import('electron')
])
return new TelegramChannelDriver({
secret: settings.secret,
allowedSenderIds: settings.allowedSenderIds,
fetch: (input, init) => net.fetch(input instanceof URL ? input.href : input, init),
onStatus
})
}
return settings.channel === 'wecom'
? new WeComChannelDriver({
botId: settings.botId,
@@ -136,6 +164,10 @@ function sanitizedManagerFailure(message: string): Error {
}
function validateResolved(settings: ResolvedChannelSettings): void {
if (settings.channel === 'telegram') {
if (!settings.secret) throw new Error('Telegram 需要 Bot Token')
return
}
if (settings.channel === 'weixin') {
if (
settings.accountId.length === 0 ||
@@ -175,7 +207,13 @@ export class ChannelManager {
private readonly createService: ChannelServiceFactory
private readonly dedupStore?: ChannelServiceOptions['dedupStore']
private readonly outbox?: ChannelServiceOptions['outbox']
private readonly onStatusChanged?: ChannelManagerOptions['onStatusChanged']
private operationQueue: Promise<void> = Promise.resolve()
private readonly serviceGenerations = new Map<ManagedChannel, symbol>()
private activeTelegram?: {
settings: Extract<ResolvedChannelSettings, { channel: 'telegram' }>
driver: TestableChannelDriver
}
constructor(
private readonly store: ChannelSettingsStore,
@@ -184,11 +222,12 @@ export class ChannelManager {
) {
this.createDriver =
options.createDriver ??
((settings) =>
defaultDriverFactory(settings, options.launchWechatSidecar))
((settings, onStatus) =>
defaultDriverFactory(settings, options.launchWechatSidecar, onStatus))
this.createService = options.createService ?? defaultServiceFactory
this.dedupStore = options.dedupStore
this.outbox = options.outbox
this.onStatusChanged = options.onStatusChanged
}
snapshot(): Promise<ChannelSettingsSnapshot> {
@@ -204,7 +243,7 @@ export class ChannelManager {
const settings = await this.store.resolveAll()
for (const channelSettings of settings) {
if (!channelSettings.enabled) {
this.statuses.set(channelSettings.channel, {
this.setStatus(channelSettings.channel, {
state: 'disabled'
})
continue
@@ -221,11 +260,17 @@ export class ChannelManager {
apply(input: ChannelSettingsApply): Promise<ChannelSettingsSnapshot> {
return this.enqueue(async () => {
input = channelSettingsApplySchema.parse(input)
if (input.telegram?.secret.action === 'replace') {
const result = await this.test('telegram', input.telegram)
if (!result.ok) throw sanitizedManagerFailure(result.error!)
}
await this.store.apply(input)
const channels: ManagedChannel[] = [
...(input.weixin === undefined ? [] : (['weixin'] as const)),
...(input.wecom === undefined ? [] : (['wecom'] as const)),
...(input.dingtalk === undefined ? [] : (['dingtalk'] as const))
...(input.dingtalk === undefined ? [] : (['dingtalk'] as const)),
...(input.telegram === undefined ? [] : (['telegram'] as const))
]
for (const channel of channels) {
const settings = await this.store.resolve(channel)
@@ -233,12 +278,27 @@ export class ChannelManager {
await this.disableService(channel)
continue
}
const active = this.activeTelegram
const service = this.services.get(channel)
if (settings.channel === 'telegram' && active &&
active.settings.secret === settings.secret &&
(this.statuses.get(channel)?.state === 'running' || this.statuses.get(channel)?.state === 'starting') &&
active.driver.updateAllowedSenderIds && service?.updateAllowedSenderIds) {
active.driver.updateAllowedSenderIds(settings.allowedSenderIds)
service.updateAllowedSenderIds(settings.allowedSenderIds)
active.settings = settings
continue
}
await this.replaceService(settings)
}
return this.snapshot()
})
}
test(
channel: 'telegram',
settings?: TelegramChannelSettingsInput
): Promise<ChannelConnectionTestResult>
test(
channel: 'wecom',
settings?: WeComChannelSettingsInput
@@ -249,7 +309,7 @@ export class ChannelManager {
): Promise<ChannelConnectionTestResult>
async test(
channel: CredentialChannel,
settings?: WeComChannelSettingsInput | DingTalkChannelSettingsInput
settings?: WeComChannelSettingsInput | DingTalkChannelSettingsInput | TelegramChannelSettingsInput
): Promise<ChannelConnectionTestResult> {
let resolved: ResolvedChannelSettings | undefined
try {
@@ -258,7 +318,21 @@ export class ChannelManager {
...(settings === undefined ? {} : { settings })
} as TestSettingsInput)
validateResolved(resolved)
const service = await this.buildService(resolved)
if (channel === 'telegram') {
const driver = await this.createDriver(resolved)
try {
if (!driver.testConnection) throw new Error('Telegram connection test is unavailable')
const identity = await driver.testConnection()
return channelConnectionTestResultSchema.parse({
channel,
ok: true,
...(identity.botUsername ? { botUsername: identity.botUsername } : {})
})
} finally {
await Promise.resolve(driver.stop()).catch(() => undefined)
}
}
const { service } = await this.buildService(resolved)
try {
await service.start()
} finally {
@@ -284,6 +358,10 @@ export class ChannelManager {
}
}
testConnection(
channel: 'telegram',
settings?: TelegramChannelSettingsInput
): Promise<ChannelConnectionTestResult>
testConnection(
channel: 'wecom',
settings?: WeComChannelSettingsInput
@@ -294,8 +372,11 @@ export class ChannelManager {
): Promise<ChannelConnectionTestResult>
testConnection(
channel: CredentialChannel,
settings?: WeComChannelSettingsInput | DingTalkChannelSettingsInput
settings?: WeComChannelSettingsInput | DingTalkChannelSettingsInput | TelegramChannelSettingsInput
): Promise<ChannelConnectionTestResult> {
if (channel === 'telegram') {
return this.test(channel, settings as TelegramChannelSettingsInput | undefined)
}
return channel === 'wecom'
? this.test(
channel,
@@ -310,13 +391,15 @@ export class ChannelManager {
stopAll(): Promise<void> {
return this.enqueue(async () => {
const active = [...this.services.entries()]
this.serviceGenerations.clear()
this.activeTelegram = undefined
this.services.clear()
const results = await Promise.allSettled(
active.map(([, service]) => Promise.resolve(service.stop()))
)
const resolved = await this.store.resolveAll()
for (const settings of resolved) {
this.statuses.set(settings.channel, {
this.setStatus(settings.channel, {
state: settings.enabled ? 'stopped' : 'disabled'
})
}
@@ -344,17 +427,24 @@ export class ChannelManager {
): Promise<void> {
const channel = settings.channel
const previous = this.services.get(channel)
this.statuses.set(channel, { state: 'starting' })
const generation = Symbol(channel)
this.serviceGenerations.set(channel, generation)
this.setStatus(channel, { state: 'starting' })
let replacement: ManagedChannelService | undefined
let driver: TestableChannelDriver | undefined
try {
validateResolved(settings)
replacement = await this.buildService(settings)
const built = await this.buildService(settings, generation)
replacement = built.service
driver = built.driver
if (previous !== undefined) {
await previous.stop()
this.services.delete(channel)
}
await replacement.start()
} catch (error) {
if (channel === 'telegram') this.activeTelegram = undefined
this.serviceGenerations.delete(channel)
await Promise.resolve(replacement?.stop()).catch(() => undefined)
if (
previous !== undefined &&
@@ -368,7 +458,7 @@ export class ChannelManager {
? settings.token
: settings.secret
])
this.statuses.set(channel, {
this.setStatus(channel, {
state: 'error',
lastError: redacted
})
@@ -376,29 +466,48 @@ export class ChannelManager {
}
this.services.set(channel, replacement)
this.statuses.set(channel, { state: 'running' })
if (settings.channel === 'telegram') {
this.activeTelegram = { settings, driver: driver! }
} else if (this.statuses.get(channel)?.state === 'starting') {
this.setStatus(channel, { state: 'running' })
}
}
private async disableService(channel: ManagedChannel): Promise<void> {
if (channel === 'telegram') this.activeTelegram = undefined
this.serviceGenerations.delete(channel)
const previous = this.services.get(channel)
if (previous !== undefined) {
await previous.stop()
this.services.delete(channel)
}
this.statuses.set(channel, { state: 'disabled' })
this.setStatus(channel, { state: 'disabled' })
}
private async buildService(
settings: ResolvedChannelSettings
): Promise<ManagedChannelService> {
const driver = await this.createDriver(settings)
return this.createService(driver, this.executor, {
settings: ResolvedChannelSettings,
generation?: symbol
): Promise<{ service: ManagedChannelService; driver: TestableChannelDriver }> {
const onStatus = (status: ChannelRuntimeStatus): void => {
if (generation === undefined || this.serviceGenerations.get(settings.channel) !== generation) return
this.setStatus(settings.channel, {
state: status.state,
...(status.lastError ? {
lastError: redactManagerError(status.lastError, [
settings.channel === 'weixin' ? settings.token : settings.secret
])
} : {})
})
}
const driver = await this.createDriver(settings, onStatus)
const service = await this.createService(driver, this.executor, {
allowedSenderIds: settings.allowedSenderIds,
allowGroupMessages: settings.allowGroupMessages,
dedupStore: this.dedupStore,
outbox: this.outbox,
onDeliveryFailure: (error) => {
this.statuses.set(settings.channel, {
if (settings.channel === 'telegram') return
onStatus({
state: 'error',
lastError: redactManagerError(error, [
settings.channel === 'weixin'
@@ -408,14 +517,28 @@ export class ChannelManager {
})
},
onDeliverySuccess: () => {
this.statuses.set(settings.channel, { state: 'running' })
if (settings.channel === 'telegram') return
onStatus({ state: 'running' })
}
})
return { service, driver }
}
private async settingsForTest(
input: TestSettingsInput
): Promise<ResolvedChannelSettings> {
if (input.channel === 'telegram') {
const current = await this.store.resolve('telegram')
if (input.settings === undefined) return current
if (current.readOnly) throw new Error('环境变量通道配置为只读,不能使用临时设置')
const parsed = telegramChannelSettingsInputSchema.parse(input.settings)
return {
channel: 'telegram',
enabled: parsed.enabled,
...this.testCommonSettings(current.secret, parsed),
allowGroupMessages: false
}
}
if (input.channel === 'wecom') {
const current = await this.store.resolve('wecom')
if (input.settings === undefined) {
@@ -450,7 +573,7 @@ export class ChannelManager {
private testCommonSettings(
currentSecret: string | undefined,
input: WeComChannelSettingsInput | DingTalkChannelSettingsInput
input: WeComChannelSettingsInput | DingTalkChannelSettingsInput | TelegramChannelSettingsInput
): {
secret?: string
allowedSenderIds: readonly string[]
@@ -473,6 +596,13 @@ export class ChannelManager {
}
}
private setStatus(channel: ManagedChannel, status: ChannelRuntimeStatus): void {
const previous = this.statuses.get(channel)
if (previous?.state === status.state && previous.lastError === status.lastError) return
this.statuses.set(channel, status)
this.onStatusChanged?.({ channel, status: { ...status } })
}
private enqueue<T>(operation: () => Promise<T>): Promise<T> {
let value!: T
const run = async (): Promise<void> => {
+283 -6
View File
@@ -1,4 +1,4 @@
import { describe, expect, it, vi } from 'vitest'
import { afterEach, describe, expect, it, vi } from 'vitest'
import {
channelInboundTextSchema,
type ChannelInboundText,
@@ -7,6 +7,7 @@ import {
import {
MemoryDedupStore,
MemoryOutbox,
isPermanentChannelError,
type ChannelDriver,
type ChannelInboundHandler
} from './channel-driver'
@@ -135,6 +136,280 @@ describe('channel contracts', () => {
})
describe('ChannelService', () => {
afterEach(() => {
vi.useRealTimers()
})
it('stops the active request without launching queued conversation requests', async () => {
const driver = new FakeChannelDriver()
const executor = vi.fn((_message: ChannelInboundText, _signal: AbortSignal) => {
void _message
void _signal
return new Promise<never>(() => undefined)
})
const service = new ChannelService(driver, executor, {
allowedSenderIds: ['allowed-user']
})
await service.start()
await driver.emit(inbound())
await driver.emit(inbound({ eventId: 'queued' }))
await service.stop()
expect(executor).toHaveBeenCalledOnce()
expect(executor.mock.calls[0]?.[1].aborted).toBe(true)
expect(driver.sent).toEqual([])
})
it('does not launch a request whose asynchronous event claim finishes after stop', async () => {
const driver = new FakeChannelDriver()
let finishClaim!: (claimed: boolean) => void
const release = vi.fn()
const executor = vi.fn()
const service = new ChannelService(driver, executor, {
allowedSenderIds: ['allowed-user'],
dedupStore: {
claim: () => new Promise<boolean>((resolve) => { finishClaim = resolve }),
release
}
})
await service.start()
await driver.emit(inbound())
const stopped = service.stop()
finishClaim(true)
await stopped
expect(executor).not.toHaveBeenCalled()
expect(release).toHaveBeenCalledWith('fake', 'default', 'event-1')
})
it('updates and validates the whitelist without cancelling active work', async () => {
const driver = new FakeChannelDriver()
let finish!: (result: { status: string }) => void
const executor = vi.fn((_message: ChannelInboundText, _signal: AbortSignal) => {
void _message
void _signal
return new Promise<{ status: string }>((resolve) => { finish = resolve })
})
const service = new ChannelService(driver, executor, {
allowedSenderIds: ['allowed-user']
})
await service.start()
await driver.emit(inbound())
await driver.emit(inbound({ eventId: 'queued' }))
service.updateAllowedSenderIds([' new-user '])
expect(() => service.updateAllowedSenderIds([' '])).toThrow('白名单')
expect(executor.mock.calls[0]?.[1].aborted).toBe(false)
finish({ status: 'completed' })
await waitForSent(driver, 1)
await driver.emit(inbound({ senderId: 'new-user', eventId: 'new-user' }))
await vi.waitFor(() => expect(executor).toHaveBeenCalledTimes(2))
finish({ status: 'completed' })
await waitForSent(driver, 2)
expect(driver.sent.map((message) => message.eventId)).toEqual(['event-1', 'new-user'])
await service.stop()
})
it('starts without waiting for a blocked retry, then cancels and awaits it on stop', async () => {
vi.useFakeTimers()
const driver = new FakeChannelDriver()
const outbox = new MemoryOutbox()
outbox.enqueue({ ...inbound(), status: 'completed', recipientId: 'allowed-user' })
let finish!: () => void
let retrySignal!: AbortSignal
const send = vi.spyOn(driver, 'send').mockImplementation((_message, signal) => {
retrySignal = signal
return new Promise<void>((resolve) => { finish = resolve })
})
const service = new ChannelService(driver, vi.fn(), { outbox })
await service.start()
await vi.advanceTimersByTimeAsync(0)
expect(send).toHaveBeenCalledOnce()
await vi.advanceTimersByTimeAsync(120_000)
expect(send).toHaveBeenCalledOnce()
let stopped = false
const stop = service.stop().then(() => { stopped = true })
await Promise.resolve()
expect(retrySignal.aborted).toBe(true)
expect(stopped).toBe(false)
finish()
await stop
expect(stopped).toBe(true)
expect(vi.getTimerCount()).toBe(0)
})
it('retries failed runtime results on recovery and does not poll an empty outbox', async () => {
vi.useFakeTimers()
const driver = new FakeChannelDriver()
const outbox = new MemoryOutbox()
const list = vi.spyOn(outbox, 'listUndelivered')
const send = vi.spyOn(driver, 'send').mockRejectedValueOnce(new Error('offline'))
const executor = vi.fn(async () => ({ status: 'failed', error: 'runtime failed' }))
const service = new ChannelService(driver, executor, {
outbox, allowedSenderIds: ['allowed-user']
})
await service.start()
await vi.advanceTimersByTimeAsync(0)
const idleReads = list.mock.calls.length
await vi.advanceTimersByTimeAsync(120_000)
expect(list).toHaveBeenCalledTimes(idleReads)
await driver.emit(inbound())
await vi.advanceTimersByTimeAsync(0)
expect(send).toHaveBeenCalledOnce()
await vi.advanceTimersByTimeAsync(999)
expect(send).toHaveBeenCalledOnce()
await vi.advanceTimersByTimeAsync(1)
expect(send).toHaveBeenCalledTimes(2)
expect(driver.sent[0]?.status).toBe('failed')
expect(outbox.listUndelivered()).toEqual([])
expect(executor).toHaveBeenCalledOnce()
expect(vi.getTimerCount()).toBe(0)
await service.stop()
})
it('backs off and stops retrying after five failed sends', async () => {
vi.useFakeTimers()
const driver = new FakeChannelDriver()
const outbox = new MemoryOutbox()
const send = vi.spyOn(driver, 'send').mockRejectedValue(new Error('offline'))
const service = new ChannelService(driver, async () => ({ status: 'completed' }), {
outbox, allowedSenderIds: ['allowed-user']
})
await service.start()
await vi.advanceTimersByTimeAsync(0)
await driver.emit(inbound())
await vi.advanceTimersByTimeAsync(0)
for (const [index, delay] of [1_000, 2_000, 4_000, 8_000].entries()) {
await vi.advanceTimersByTimeAsync(delay - 1)
expect(send).toHaveBeenCalledTimes(index + 1)
await vi.advanceTimersByTimeAsync(1)
expect(send).toHaveBeenCalledTimes(index + 2)
}
expect(outbox.listUndelivered()[0]).toMatchObject({ state: 'terminal', attempts: 5 })
expect(vi.getTimerCount()).toBe(0)
await vi.advanceTimersByTimeAsync(120_000)
expect(send).toHaveBeenCalledTimes(5)
await service.stop()
})
it('bounds recovery passes and backs off after three consecutive failures', async () => {
vi.useFakeTimers()
const driver = new FakeChannelDriver()
const outbox = new MemoryOutbox()
for (let index = 0; index < 101; index += 1) {
outbox.enqueue({
channel: 'fake', eventId: `saved-${index}`, conversationId: 'conversation-1',
recipientId: 'allowed-user', status: 'completed'
})
}
const send = vi.spyOn(driver, 'send').mockRejectedValue(new Error('offline'))
const service = new ChannelService(driver, vi.fn(), { outbox })
await service.start()
await vi.advanceTimersByTimeAsync(0)
expect(send).toHaveBeenCalledTimes(3)
send.mockResolvedValue(undefined)
await vi.advanceTimersByTimeAsync(1_999)
expect(send).toHaveBeenCalledTimes(3)
await vi.advanceTimersByTimeAsync(1)
expect(send).toHaveBeenCalledTimes(103)
expect(outbox.listUndelivered()).toHaveLength(1)
await vi.advanceTimersByTimeAsync(1_000)
expect(send).toHaveBeenCalledTimes(104)
expect(outbox.listUndelivered()).toEqual([])
expect(vi.getTimerCount()).toBe(0)
await service.stop()
})
it('does not retry an in-flight live delivery and aborts its send on stop', async () => {
vi.useFakeTimers()
const driver = new FakeChannelDriver()
let liveSignal!: AbortSignal
const send = vi.spyOn(driver, 'send').mockImplementation((message, signal) => {
if (message.eventId === 'held') {
liveSignal = signal
return new Promise<void>((_resolve, reject) => {
signal.addEventListener('abort', () => reject(signal.reason), { once: true })
})
}
return Promise.reject(new Error('offline'))
})
const service = new ChannelService(driver, async () => ({ status: 'completed' }), {
allowedSenderIds: ['allowed-user']
})
await service.start()
await vi.advanceTimersByTimeAsync(0)
await driver.emit(inbound({ eventId: 'held' }))
await driver.emit(inbound({ eventId: 'failed', conversationId: 'another' }))
await vi.advanceTimersByTimeAsync(1_000)
expect(send.mock.calls.filter(([message]) => message.eventId === 'held')).toHaveLength(1)
expect(send.mock.calls.filter(([message]) => message.eventId === 'failed')).toHaveLength(2)
await service.stop()
expect(liveSignal.aborted).toBe(true)
expect(vi.getTimerCount()).toBe(0)
})
it('excludes live deliveries from a retry snapshot even if they finish before the retry reaches them', async () => {
vi.useFakeTimers()
const driver = new FakeChannelDriver()
const outbox = new MemoryOutbox()
let finishLive!: () => void
let finishRetry!: () => void
const send = vi.spyOn(driver, 'send').mockImplementation((message) =>
new Promise<void>((resolve) => {
if (message.eventId === 'live') finishLive = resolve
else finishRetry = resolve
})
)
const service = new ChannelService(driver, async () => ({ status: 'completed' }), {
outbox, allowedSenderIds: ['allowed-user']
})
await service.start()
await vi.advanceTimersByTimeAsync(0)
outbox.enqueue({ ...inbound({ eventId: 'retry' }), recipientId: 'allowed-user', status: 'completed' })
await driver.emit(inbound({ eventId: 'live' }))
await vi.advanceTimersByTimeAsync(0)
// A failed delivery schedules the recovery pass through the public path.
send.mockRejectedValueOnce(new Error('offline'))
await driver.emit(inbound({ eventId: 'failed', conversationId: 'another' }))
await vi.advanceTimersByTimeAsync(1_000)
expect(send.mock.calls.map(([message]) => message.eventId)).toEqual(['live', 'failed', 'retry'])
finishLive()
await vi.advanceTimersByTimeAsync(0)
expect(outbox.listUndelivered().some((entry) => entry.message.eventId === 'live')).toBe(false)
send.mockResolvedValue(undefined)
finishRetry()
await vi.advanceTimersByTimeAsync(0)
expect(send.mock.calls.filter(([message]) => message.eventId === 'live')).toHaveLength(1)
expect(outbox.listUndelivered()).toEqual([])
await service.stop()
})
it.each(['field', 'getter'] as const)('terminates permanent %s errors without resending after restart', async (kind) => {
vi.useFakeTimers()
const error = kind === 'field'
? Object.assign(new Error('forbidden'), { permanent: true })
: new class extends Error { get permanent(): boolean { return true } }('forbidden')
expect(isPermanentChannelError(error)).toBe(true)
expect(isPermanentChannelError(new Error('offline'))).toBe(false)
const driver = new FakeChannelDriver()
const outbox = new MemoryOutbox()
const send = vi.spyOn(driver, 'send').mockRejectedValue(error)
const service = new ChannelService(driver, async () => ({
status: 'completed',
attachments: [{ name: 'result.txt', mimeType: 'text/plain', size: 2, kind: 'file', dataBase64: 'b2s=' }]
}), { outbox, allowedSenderIds: ['allowed-user'] })
await service.start()
await vi.advanceTimersByTimeAsync(0)
await driver.emit(inbound())
await vi.advanceTimersByTimeAsync(0)
expect(outbox.listUndelivered()[0]).toMatchObject({ state: 'terminal', attempts: 1 })
expect(outbox.listUndelivered()[0]?.message.attachments).toBeUndefined()
await service.stop()
const restarted = new ChannelService(driver, vi.fn(), { outbox })
await restarted.start()
await vi.advanceTimersByTimeAsync(120_000)
expect(send).toHaveBeenCalledOnce()
expect(vi.getTimerCount()).toBe(0)
await restarted.stop()
})
it('acknowledges after accepting input and denies all senders when no allowlist is configured', async () => {
const driver = new FakeChannelDriver()
const executor = vi.fn()
@@ -462,11 +737,13 @@ describe('ChannelService', () => {
await service.start()
expect(driver.sent).toEqual([])
expect(deliveryFailure).toHaveBeenCalledWith(
expect.objectContaining({
message: expect.stringContaining('已达到重试上限')
})
)
await vi.waitFor(() => {
expect(deliveryFailure).toHaveBeenCalledWith(
expect.objectContaining({
message: expect.stringContaining('已达到重试上限')
})
)
})
expect(await outbox.listUndelivered()).toEqual([
expect.objectContaining({
id: entry.id,
+123 -25
View File
@@ -9,6 +9,7 @@ import {
import {
MemoryDedupStore,
MemoryOutbox,
isPermanentChannelError,
type ChannelDriver,
type ChannelExecutor,
type DedupStore,
@@ -64,7 +65,7 @@ export function redactChannelError(value: string): string {
}
export class ChannelService {
private readonly allowedSenderIds: ReadonlySet<string>
private allowedSenderIds: ReadonlySet<string> = new Set()
private readonly allowGroupMessages: boolean
private readonly maximumConcurrency: number
private readonly maximumInputLength: number
@@ -76,6 +77,12 @@ export class ChannelService {
private readonly conversationTails = new Map<string, Promise<void>>()
private state: ServiceState = 'idle'
private stopPromise?: Promise<void>
private readonly lifetime = new AbortController()
private readonly sending = new Set<string>()
private retryTimer?: ReturnType<typeof setTimeout>
private retryPromise?: Promise<void>
private retryRequested = false
private retryDelay = 1_000
constructor(
private readonly driver: ChannelDriver,
@@ -90,12 +97,7 @@ export class ChannelService {
throw new Error('通道标识无效')
}
this.allowedSenderIds = new Set(
(options.allowedSenderIds ?? []).map((senderId) => senderId.trim())
)
if (this.allowedSenderIds.has('')) {
throw new Error('通道白名单包含无效身份')
}
this.updateAllowedSenderIds(options.allowedSenderIds ?? [])
this.allowGroupMessages = options.allowGroupMessages ?? false
this.maximumConcurrency = boundedInteger(
options.maximumConcurrency,
@@ -115,6 +117,14 @@ export class ChannelService {
this.onDeliverySuccess = options.onDeliverySuccess
}
updateAllowedSenderIds(senderIds: readonly string[]): void {
const allowed = new Set(senderIds.map((senderId) => senderId.trim()))
if (allowed.has('')) {
throw new Error('通道白名单包含无效身份')
}
this.allowedSenderIds = allowed
}
async start(): Promise<void> {
if (this.state === 'running') {
return
@@ -138,9 +148,11 @@ export class ChannelService {
this.onDeliveryFailure?.(error)
}
})
await this.retryUndelivered()
this.scheduleRetry(0)
} catch (error) {
this.state = 'idle'
if (!this.lifetime.signal.aborted) {
this.state = 'idle'
}
throw error
}
}
@@ -168,6 +180,9 @@ export class ChannelService {
}
this.state = 'stopped'
this.lifetime.abort(new Error('通道服务已停止'))
clearTimeout(this.retryTimer)
this.retryTimer = undefined
for (const controller of this.active.values()) {
controller.abort(new Error('通道服务已停止'))
}
@@ -180,6 +195,7 @@ export class ChannelService {
const driverStop = Promise.resolve().then(() => this.driver.stop())
const results = await Promise.allSettled([
driverStop,
this.retryPromise,
...this.conversationTails.values()
])
const driverResult = results[0]
@@ -188,12 +204,52 @@ export class ChannelService {
}
}
private scheduleRetry(delay = this.retryDelay): void {
if (this.state !== 'running') {
return
}
if (this.retryPromise) {
this.retryRequested = true
return
}
if (this.retryTimer) {
return
}
const run = (): void => {
this.retryTimer = undefined
this.retryRequested = false
this.retryPromise = this.retryUndelivered()
.catch((error: unknown) => {
if (!this.lifetime.signal.aborted) {
this.onDeliveryFailure?.(error)
}
})
.finally(() => {
this.retryPromise = undefined
if (this.retryRequested) {
this.scheduleRetry()
} else {
this.retryDelay = 1_000
}
})
}
if (delay === 0) {
run()
} else {
this.retryTimer = setTimeout(run, delay)
this.retryTimer.unref?.()
}
}
private async retryUndelivered(): Promise<void> {
const entries = await this.outbox.listUndelivered(
// Live sends may finish while this snapshot is read or earlier retries await I/O.
const liveDeliveries = new Set(this.sending)
const entries = (await this.outbox.listUndelivered(
this.driver.channel,
100
)
)).filter((entry) => !liveDeliveries.has(entry.id) && !this.sending.has(entry.id))
let consecutiveFailures = 0
let failed = false
for (const entry of entries) {
if (this.state !== 'running') {
return
@@ -201,31 +257,49 @@ export class ChannelService {
if (entry.state === 'terminal' || entry.attempts >= 5) {
this.onDeliveryFailure?.(
new Error(
`通道结果已达到重试上限,发件箱记录 ${entry.id} 已终止`
entry.attempts >= 5
? `通道结果已达到重试上限,发件箱记录 ${entry.id} 已终止`
: `通道结果无法投递,发件箱记录 ${entry.id} 已终止`
)
)
continue
}
if (this.sending.has(entry.id)) {
continue
}
try {
await this.driver.send(
entry.message,
new AbortController().signal
this.lifetime.signal
)
await this.outbox.markDelivered(entry.id)
this.onDeliverySuccess?.()
consecutiveFailures = 0
} catch (error) {
await this.outbox.markFailed(entry.id)
this.onDeliveryFailure?.(error)
consecutiveFailures += 1
if (consecutiveFailures >= 3) {
if (this.lifetime.signal.aborted) {
return
}
await this.recordDeliveryFailure(entry.id, error)
failed = true
consecutiveFailures += 1
if (consecutiveFailures >= 3) {
break
}
}
}
if (this.state === 'running') {
const remaining = await this.outbox.listUndelivered(this.driver.channel, 100)
this.retryRequested ||= remaining.some((entry) =>
entry.state !== 'terminal' && entry.attempts < 5 && !this.sending.has(entry.id)
)
this.retryDelay = failed ? Math.min(this.retryDelay * 2, 30_000) : 1_000
}
}
private async process(rawMessage: unknown): Promise<void> {
if (this.state !== 'running') {
return
}
const parsed = channelInboundTextSchema.safeParse(rawMessage)
if (!parsed.success) {
return
@@ -252,13 +326,16 @@ export class ChannelService {
let durableResult = false
try {
if (this.lifetime.signal.aborted || !this.allowedSenderIds.has(message.senderId)) {
return
}
if (message.text.length > this.maximumInputLength) {
durableResult = await this.tryDeliver(
this.result(message, {
status: 'rejected',
error: `消息过长,最多允许 ${this.maximumInputLength} 个字符`
}),
new AbortController().signal
this.lifetime.signal
)
return
}
@@ -269,7 +346,7 @@ export class ChannelService {
status: 'busy',
error: '当前请求较多,请稍后重试'
}),
new AbortController().signal
this.lifetime.signal
)
return
}
@@ -292,7 +369,7 @@ export class ChannelService {
status: cancelled ? 'cancelled' : 'failed',
error: cancelled ? '请求已取消' : '请求处理失败'
}),
new AbortController().signal
this.lifetime.signal
)
return
}
@@ -302,7 +379,7 @@ export class ChannelService {
status: 'cancelled',
error: '请求已取消'
}),
new AbortController().signal
this.lifetime.signal
)
return
}
@@ -382,7 +459,10 @@ export class ChannelService {
)
}
void Promise.resolve()
.then(() => this.executor(message, signal, reportProgress))
.then(() => {
signal.throwIfAborted()
return this.executor(message, signal, reportProgress)
})
.then(
(result) => finish(resolve, result),
(error: unknown) => finish(reject, error)
@@ -429,17 +509,35 @@ export class ChannelService {
signal: AbortSignal
): Promise<boolean> {
const entry = await this.outbox.enqueue(message)
this.sending.add(entry.id)
try {
await this.driver.send(message, signal)
const deliverySignal = AbortSignal.any([signal, this.lifetime.signal])
deliverySignal.throwIfAborted()
await this.driver.send(message, deliverySignal)
await this.outbox.markDelivered(entry.id)
this.onDeliverySuccess?.()
} catch (error) {
await this.outbox.markFailed(entry.id)
this.onDeliveryFailure?.(error)
if (!this.lifetime.signal.aborted) {
await this.recordDeliveryFailure(entry.id, error)
if (!isPermanentChannelError(error)) {
this.scheduleRetry()
}
}
} finally {
this.sending.delete(entry.id)
}
return true
}
private async recordDeliveryFailure(id: string, error: unknown): Promise<void> {
if (isPermanentChannelError(error) && this.outbox.markTerminal) {
await this.outbox.markTerminal(id)
} else {
await this.outbox.markFailed(id)
}
this.onDeliveryFailure?.(error)
}
private async tryDeliver(
message: ChannelResultMessage,
signal: AbortSignal
@@ -40,6 +40,75 @@ function createCipher(available = true): ChannelCredentialCipher {
}
describe('ChannelSettingsStore', () => {
it('persists encrypted Telegram settings with empty whitelist and secret actions', async () => {
const filePath = await settingsPath()
const store = new ChannelSettingsStore(filePath, createCipher(), {})
const input = { enabled: true, allowedSenderIds: [], allowGroupMessages: false as const }
const snapshot = await store.apply({ telegram: {
...input, secret: { action: 'replace', value: 'test-only-secret' }
} })
expect(snapshot.telegram).toMatchObject({ enabled: true, secretConfigured: true, source: 'encrypted' })
expect(JSON.stringify(snapshot)).not.toContain('test-only-secret')
expect(snapshot.telegram).not.toHaveProperty('botId')
expect(await readFile(filePath, 'utf8')).not.toContain('test-only-secret')
const reloaded = new ChannelSettingsStore(filePath, createCipher(), {})
expect(await reloaded.resolve('telegram')).toEqual({
channel: 'telegram', ...input, secret: 'test-only-secret', source: 'encrypted', readOnly: false
})
await reloaded.apply({ telegram: { ...input, secret: { action: 'keep' }, allowedSenderIds: ['123'] } })
expect((await reloaded.resolve('telegram')).secret).toBe('test-only-secret')
await reloaded.apply({ telegram: { ...input, secret: { action: 'replace', value: 'replacement-test-secret' } } })
expect((await reloaded.resolve('telegram')).secret).toBe('replacement-test-secret')
const unreadable = new ChannelSettingsStore(filePath, createCipher(false), {})
expect((await unreadable.snapshot()).telegram).toMatchObject({ source: 'unreadable', secretConfigured: false })
await reloaded.apply({ telegram: { ...input, enabled: false, secret: { action: 'clear' } } })
expect((await reloaded.snapshot()).telegram).toMatchObject({ enabled: false, source: 'none', secretConfigured: false })
await expect(reloaded.apply({ telegram: { ...input, secret: { action: 'keep' } } })).rejects.toThrow('Bot Token')
})
it('loads existing version 3 settings with Telegram defaults without rewriting', async () => {
const filePath = await settingsPath()
const original = JSON.stringify({
version: 3,
weixin: { enabled: false },
wecom: { enabled: false, botId: '', allowedSenderIds: [], allowGroupMessages: false },
dingtalk: { enabled: false, clientId: '', allowedSenderIds: [], allowGroupMessages: false }
})
await writeFile(filePath, original)
const store = new ChannelSettingsStore(filePath, createCipher(), {})
expect((await store.snapshot()).telegram).toEqual({
enabled: false, secretConfigured: false, source: 'none', readOnly: false,
allowedSenderIds: [], allowGroupMessages: false, status: { state: 'disabled' }
})
expect(await readFile(filePath, 'utf8')).toBe(original)
})
it('uses read-only Telegram environment credentials, including help-only startup', async () => {
const filePath = await settingsPath()
const environment = { GOODBUDDY_TELEGRAM_BOT_TOKEN: 'environment-test-secret' }
const store = new ChannelSettingsStore(filePath, createCipher(), environment)
expect(await store.resolve('telegram')).toEqual({
channel: 'telegram', enabled: true, secret: 'environment-test-secret',
allowedSenderIds: [], allowGroupMessages: false, source: 'environment', readOnly: true
})
await expect(store.apply({ telegram: {
enabled: false, secret: { action: 'clear' }, allowedSenderIds: [], allowGroupMessages: false
} })).rejects.toThrow('环境变量配置')
const configured = new ChannelSettingsStore(filePath, createCipher(), {
...environment, GOODBUDDY_TELEGRAM_ALLOWED_SENDERS: '123, 456,123', GOODBUDDY_TELEGRAM_ENABLED: 'false'
})
expect(await configured.resolve('telegram')).toMatchObject({ enabled: false, allowedSenderIds: ['123', '456'] })
for (const invalid of [
{ GOODBUDDY_TELEGRAM_ALLOWED_SENDERS: '@user' },
{ GOODBUDDY_TELEGRAM_ALLOW_GROUPS: 'true' },
{ GOODBUDDY_TELEGRAM_ENABLED: 'invalid' }
]) {
const invalidStore = new ChannelSettingsStore(filePath, createCipher(), { ...environment, ...invalid })
expect((await invalidStore.snapshot()).telegram).toMatchObject({
enabled: false, readOnly: true, allowGroupMessages: false, status: { state: 'error' }
})
}
})
it('encrypts secrets and supports keep, replace, and clear', async () => {
const filePath = await settingsPath()
const store = new ChannelSettingsStore(filePath, createCipher(), {})
+111 -11
View File
@@ -3,6 +3,7 @@ import { z } from 'zod'
import {
CHANNEL_SETTINGS_LIMITS,
allowedSenderIdsSchema,
telegramAllowedSenderIdsSchema,
channelSettingsApplySchema,
type ChannelRuntimeStatus,
type ChannelSettingsApply,
@@ -10,6 +11,7 @@ import {
type CredentialChannel,
type DingTalkChannelSettingsInput,
type ManagedChannel,
type TelegramChannelSettingsInput,
type WeComChannelSettingsInput
} from '../../shared/channel-settings-contracts'
import { weixinAccountDisplay } from '../../shared/weixin-channel-contracts'
@@ -103,7 +105,16 @@ const storedSettingsSchema = z
})
.strict(),
wecom: legacyStoredSettingsSchema.shape.wecom,
dingtalk: legacyStoredSettingsSchema.shape.dingtalk
dingtalk: legacyStoredSettingsSchema.shape.dingtalk,
telegram: z.object({
...storedChannelFields,
allowedSenderIds: telegramAllowedSenderIdsSchema,
allowGroupMessages: z.literal(false)
}).strict().default({
enabled: false,
allowedSenderIds: [],
allowGroupMessages: false
})
})
.strict()
@@ -111,6 +122,7 @@ type StoredSettings = z.infer<typeof storedSettingsSchema>
type StoredCredentialChannel =
| StoredSettings['wecom']
| StoredSettings['dingtalk']
| StoredSettings['telegram']
type StoredEncryptedCredential = z.infer<
typeof encryptedCredentialSchema
>
@@ -120,7 +132,7 @@ class DeferredWeixinMigrationError extends Error {}
const credentialPayloadSchema = z
.object({
version: z.literal(1),
channel: z.enum(['weixin', 'wecom', 'dingtalk']),
channel: z.enum(['weixin', 'wecom', 'dingtalk', 'telegram']),
secret: z
.string()
.min(1)
@@ -167,9 +179,19 @@ type EnvironmentChannel = {
allowedSenderIds: readonly string[]
allowGroupMessages: boolean
warning?: SettingsWarning
configurationError?: string
}
export type ResolvedChannelSettings =
| {
channel: 'telegram'
enabled: boolean
secret?: string
allowedSenderIds: readonly string[]
allowGroupMessages: false
source: 'none' | 'encrypted' | 'environment' | 'unreadable'
readOnly: boolean
}
| {
channel: 'weixin'
enabled: boolean
@@ -205,6 +227,11 @@ export type ResolvedChannelSettings =
const defaultStoredSettings: StoredSettings = {
version: 3,
telegram: {
enabled: false,
allowedSenderIds: [],
allowGroupMessages: false
},
weixin: {
enabled: false
},
@@ -337,20 +364,23 @@ export class ChannelSettingsStore {
) {
this.environmentChannels = {
wecom: this.readEnvironmentChannel('wecom'),
dingtalk: this.readEnvironmentChannel('dingtalk')
dingtalk: this.readEnvironmentChannel('dingtalk'),
telegram: this.readEnvironmentChannel('telegram')
}
}
async snapshot(
statuses: Partial<Record<ManagedChannel, ChannelRuntimeStatus>> = {}
): Promise<ChannelSettingsSnapshot> {
const [weixin, wecom, dingtalk] = await Promise.all([
const [weixin, wecom, dingtalk, telegram] = await Promise.all([
this.resolve('weixin'),
this.resolve('wecom'),
this.resolve('dingtalk')
this.resolve('dingtalk'),
this.resolve('telegram')
])
const weComEnvironment = this.environmentChannel('wecom')
const dingTalkEnvironment = this.environmentChannel('dingtalk')
const telegramEnvironment = this.environmentChannel('telegram')
const warnings = [
...this.warnings,
...(this.runtimeRepairWarning ? [this.runtimeRepairWarning] : []),
@@ -363,6 +393,17 @@ export class ChannelSettingsStore {
) === index
)
return {
telegram: {
enabled: telegram.enabled,
secretConfigured: telegram.secret !== undefined,
source: telegram.source,
readOnly: telegram.readOnly,
allowedSenderIds: [...telegram.allowedSenderIds],
allowGroupMessages: false,
status: telegramEnvironment.configurationError
? { state: 'error', lastError: telegramEnvironment.configurationError }
: statuses.telegram ?? defaultStatus(telegram.enabled)
},
weixin: {
enabled: weixin.enabled,
bindingConfigured: weixin.token !== undefined,
@@ -418,6 +459,9 @@ export class ChannelSettingsStore {
return this.snapshot(statuses)
}
resolve(channel: 'telegram'): Promise<Extract<ResolvedChannelSettings, {
channel: 'telegram'
}>>
resolve(channel: 'wecom'): Promise<Extract<ResolvedChannelSettings, {
channel: 'wecom'
}>>
@@ -463,6 +507,9 @@ export class ChannelSettingsStore {
source: 'environment' as const,
readOnly: true
}
if (channel === 'telegram') {
return { channel, ...common, allowGroupMessages: false }
}
return channel === 'wecom'
? {
channel,
@@ -493,6 +540,9 @@ export class ChannelSettingsStore {
: ('encrypted' as const),
readOnly: false
}
if (channel === 'telegram') {
return { channel, ...common, allowGroupMessages: false }
}
return channel === 'wecom'
? { channel, botId: settings.wecom.botId, ...common }
: { channel, clientId: settings.dingtalk.clientId, ...common }
@@ -501,12 +551,14 @@ export class ChannelSettingsStore {
resolveAll(): Promise<readonly [
Extract<ResolvedChannelSettings, { channel: 'weixin' }>,
Extract<ResolvedChannelSettings, { channel: 'wecom' }>,
Extract<ResolvedChannelSettings, { channel: 'dingtalk' }>
Extract<ResolvedChannelSettings, { channel: 'dingtalk' }>,
Extract<ResolvedChannelSettings, { channel: 'telegram' }>
]> {
return Promise.all([
this.resolve('weixin'),
this.resolve('wecom'),
this.resolve('dingtalk')
this.resolve('dingtalk'),
this.resolve('telegram')
])
}
@@ -583,6 +635,12 @@ export class ChannelSettingsStore {
if (input.weixin !== undefined) {
current.weixin.enabled = input.weixin.enabled
}
if (input.telegram !== undefined) {
if (this.environmentChannel('telegram').owned) {
throw new Error('Telegram 由环境变量配置,不能在设置中修改')
}
current.telegram = this.updateStoredChannel('telegram', current.telegram, input.telegram)
}
if (input.wecom !== undefined) {
if (this.environmentChannel('wecom').owned) {
throw new Error('企业微信由环境变量配置,不能在设置中修改')
@@ -609,6 +667,7 @@ export class ChannelSettingsStore {
}
this.validateEnabledCredentialChannel('wecom', current.wecom)
this.validateEnabledCredentialChannel('dingtalk', current.dingtalk)
this.validateEnabledCredentialChannel('telegram', current.telegram)
await this.persist(current)
this.settings = current
if (!this.temporarilyDisabledWeixin) {
@@ -631,6 +690,11 @@ export class ChannelSettingsStore {
return this.snapshot()
}
private updateStoredChannel(
channel: 'telegram',
current: StoredSettings['telegram'],
input: TelegramChannelSettingsInput
): StoredSettings['telegram']
private updateStoredChannel(
channel: 'wecom',
current: StoredSettings['wecom'],
@@ -644,7 +708,7 @@ export class ChannelSettingsStore {
private updateStoredChannel(
channel: CredentialChannel,
current: StoredCredentialChannel,
input: WeComChannelSettingsInput | DingTalkChannelSettingsInput
input: WeComChannelSettingsInput | DingTalkChannelSettingsInput | TelegramChannelSettingsInput
): StoredCredentialChannel {
const credential =
input.secret.action === 'keep'
@@ -662,6 +726,9 @@ export class ChannelSettingsStore {
allowedSenderIds,
allowGroupMessages: input.allowGroupMessages
}
if (channel === 'telegram') {
return { ...common, allowGroupMessages: false }
}
return channel === 'wecom'
? {
...common,
@@ -693,6 +760,12 @@ export class ChannelSettingsStore {
if (!stored.enabled) {
return
}
if (channel === 'telegram') {
if (this.decryptCredential(channel, stored) === undefined) {
throw new Error('启用 Telegram 前需要配置 Bot Token')
}
return
}
const identifier =
channel === 'wecom'
? (stored as StoredSettings['wecom']).botId
@@ -732,6 +805,7 @@ export class ChannelSettingsStore {
return undefined
}
const warn = (): undefined => {
if (channel === 'telegram') return undefined
this.addWarning({
code:
channel === 'wecom'
@@ -750,7 +824,7 @@ export class ChannelSettingsStore {
if (payload.channel !== channel) {
return warn()
}
this.removeWarnings([
if (channel !== 'telegram') this.removeWarnings([
channel === 'wecom'
? 'channel-wecom-credential-unreadable'
: 'channel-dingtalk-credential-unreadable'
@@ -825,7 +899,8 @@ export class ChannelSettingsStore {
enabled: false
},
wecom: legacy.wecom,
dingtalk: legacy.dingtalk
dingtalk: legacy.dingtalk,
telegram: structuredClone(defaultStoredSettings.telegram)
}
}
await this.persist(this.settings)
@@ -924,7 +999,8 @@ export class ChannelSettingsStore {
: {})
},
wecom: settings.wecom,
dingtalk: settings.dingtalk
dingtalk: settings.dingtalk,
telegram: structuredClone(defaultStoredSettings.telegram)
}
}
@@ -939,6 +1015,30 @@ export class ChannelSettingsStore {
private readEnvironmentChannel(
channel: CredentialChannel
): EnvironmentChannel {
if (channel === 'telegram') {
const secret = boundedEnvironmentValue(
this.environment, 'GOODBUDDY_TELEGRAM_BOT_TOKEN',
CHANNEL_SETTINGS_LIMITS.maximumSecretLength
)
const owned = secret.value !== undefined || secret.invalid
const enabled = environmentBoolean(this.environment, 'GOODBUDDY_TELEGRAM_ENABLED', true)
const senders = telegramAllowedSenderIdsSchema.safeParse(
this.environment.GOODBUDDY_TELEGRAM_ALLOWED_SENDERS?.trim()
? this.environment.GOODBUDDY_TELEGRAM_ALLOWED_SENDERS.split(',')
: []
)
const groups = environmentBoolean(this.environment, 'GOODBUDDY_TELEGRAM_ALLOW_GROUPS', false)
const invalid = secret.invalid || enabled.invalid || !senders.success || groups.invalid || groups.value
return {
owned,
enabled: owned && !invalid && enabled.value,
id: '',
secret: secret.value,
allowedSenderIds: senders.success ? senders.data : [],
allowGroupMessages: false,
...(owned && invalid ? { configurationError: 'Telegram 环境变量配置无效,请检查 Token、数字用户 ID 和私聊设置' } : {})
}
}
const prefix =
channel === 'wecom' ? 'GOODBUDDY_WECOM' : 'GOODBUDDY_DINGTALK'
const idName =
@@ -55,7 +55,8 @@ describe('parseRemoteChannelPrompt', () => {
expect(projectChannelLabels).toEqual({
weixin: '微信 ClawBot',
wecom: '企业微信',
dingtalk: '钉钉'
dingtalk: '钉钉',
telegram: 'Telegram'
})
})
})
@@ -0,0 +1,72 @@
import { afterEach, describe, expect, it } from 'vitest'
import { AssistantDatabase } from '../assistant/assistant-database'
import { SqliteChannelOutbox } from './sqlite-channel-state'
const databases: AssistantDatabase[] = []
afterEach(() => {
for (const database of databases.splice(0)) database.close()
})
function createOutbox(): SqliteChannelOutbox {
const database = new AssistantDatabase(':memory:')
databases.push(database)
database.initialize('C:\\Workspace')
return new SqliteChannelOutbox(database)
}
const message = {
channel: 'telegram' as const,
eventId: 'event-1',
conversationId: 'chat-1',
recipientId: 'user-1',
status: 'completed' as const,
output: 'Result',
attachments: [{ name: 'result.txt', mimeType: 'text/plain', size: 1, kind: 'file' as const, dataBase64: 'eA==' }]
}
describe('SqliteChannelOutbox', () => {
it('marks a permanent failure terminal, releases media and counts the attempt once', () => {
const outbox = createOutbox()
const entry = outbox.enqueue(message)
outbox.markTerminal(entry.id)
outbox.markTerminal(entry.id)
outbox.markFailed(entry.id)
outbox.markDelivered(entry.id)
expect(outbox.listUndelivered()).toEqual([{
...entry, state: 'terminal', attempts: 1,
message: { ...message, attachments: undefined }
}])
expect(outbox.listUndelivered()[0]!.message).not.toHaveProperty('attachments')
})
it('keeps retryable entries ahead of terminal entries under channel and global limits', () => {
const outbox = createOutbox()
const terminal = outbox.enqueue(message)
outbox.markTerminal(terminal.id)
const retry = outbox.enqueue({ ...message, eventId: 'retry' })
outbox.markFailed(retry.id)
outbox.markFailed(retry.id)
for (const channel of [undefined, 'telegram']) {
expect(outbox.listUndelivered(channel, 1).map(entry => entry.id)).toEqual([retry.id])
expect(outbox.listUndelivered(channel).map(entry => entry.id)).toEqual([retry.id, terminal.id])
}
expect(outbox.listUndelivered('weixin')).toEqual([])
})
it('budgets retry bytes after putting terminal results last in both windows', () => {
const outbox = createOutbox()
// Retained terminal text alone exceeds the 20 MiB retry window.
for (let index = 0; index < 22; index++) {
const entry = outbox.enqueue({ ...message, eventId: `terminal-${index}`, output: 'x'.repeat(1024 * 1024) })
outbox.markTerminal(entry.id)
}
const retry = outbox.enqueue({ ...message, eventId: 'retry' })
outbox.markFailed(retry.id)
outbox.markFailed(retry.id)
for (const channel of [undefined, 'telegram']) {
const entries = outbox.listUndelivered(channel)
expect(entries[0]!.id).toBe(retry.id)
expect(entries.length).toBeLessThan(23)
}
})
})
@@ -33,6 +33,10 @@ export class SqliteChannelOutbox implements Outbox {
this.database.markChannelResult(id, 'failed')
}
markTerminal(id: string): void {
this.database.markChannelResult(id, 'terminal')
}
listUndelivered(
channel?: string,
limit?: number
@@ -0,0 +1,387 @@
import { afterEach, describe, expect, it, vi } from 'vitest'
import { channelInboundTextSchema, type ChannelResultMessage } from '../../shared/channel-contracts'
import { TelegramChannelDriver } from './telegram-channel-driver'
import { ChannelService } from './channel-service'
import { MemoryOutbox } from './channel-driver'
const netFetch = vi.hoisted(() => vi.fn())
vi.mock('electron', () => ({ net: { fetch: netFetch } }))
const secret = 'test-only-placeholder'
const drivers: TelegramChannelDriver[] = []
const ok = (result: unknown) => Response.json({ ok: true, result })
const error = (code: number, retryAfter = 2) => Response.json({ ok: false, error_code: code, parameters: { retry_after: retryAfter }, description: `https://api.telegram.org/bot${secret}/getUpdates` }, { status: code })
const pending = (signal: AbortSignal) => new Promise<Response>((_resolve, reject) => {
if (signal.aborted) reject(new Error(secret))
else signal.addEventListener('abort', () => reject(new Error(secret)), { once: true })
})
const update = (id: number, text = 'hello', extra: Record<string, unknown> = {}) => ({
update_id: id,
message: { message_id: id + 100, date: 1234, from: { id: 7, is_bot: false }, chat: { id: 7, type: 'private' }, text, ...extra }
})
function setup(allowedSenderIds: string[] = ['7']) {
const calls: { method: string; body: Record<string, unknown>; signal: AbortSignal }[] = []
const polls: ((signal: AbortSignal) => Response | Promise<Response>)[] = []
const sends: (() => Response | Promise<Response>)[] = []
const onStatus = vi.fn()
const fetch = vi.fn<typeof globalThis.fetch>(async (input, init) => {
const method = String(input).split('/').at(-1)!
const signal = init!.signal!
calls.push({ method, body: JSON.parse(String(init!.body)), signal })
if (method === 'getMe') return ok({ id: 42, is_bot: true, username: 'example_bot' })
if (method === 'getWebhookInfo') return ok({ url: '' })
if (method === 'getUpdates') return polls.shift()?.(signal) ?? pending(signal)
if (method === 'sendMessage') return sends.shift()?.() ?? ok({ message_id: 99 })
return ok(true)
})
const driver = new TelegramChannelDriver({ secret, allowedSenderIds, fetch, onStatus })
drivers.push(driver)
return { driver, calls, polls, sends, fetch, onStatus }
}
afterEach(async () => {
await Promise.all(drivers.splice(0).map((driver) => driver.stop()))
vi.useRealTimers()
})
const result = (overrides: Partial<ChannelResultMessage> = {}): ChannelResultMessage => ({
channel: 'telegram', eventId: '1', conversationId: '42:7', recipientId: '7', status: 'completed', output: 'answer', ...overrides
})
describe('TelegramChannelDriver', () => {
it('is lazy, validates with getMe/getWebhookInfo only, and never polls during repeated tests', async () => {
const { driver, calls } = setup()
expect(calls).toEqual([])
expect(await driver.testConnection()).toEqual({ botId: '42', botUsername: 'example_bot' })
await driver.testConnection()
expect(calls.map((call) => call.method)).toEqual(['getMe', 'getWebhookInfo', 'getMe', 'getWebhookInfo'])
})
it('refuses a webhook without deleting it or polling', async () => {
const { driver, fetch, calls } = setup()
fetch.mockResolvedValueOnce(ok({ id: 42, is_bot: true })).mockResolvedValueOnce(ok({ url: 'https://example.invalid/hook' }))
await expect(driver.testConnection()).rejects.toThrow('webhook')
expect(calls).toHaveLength(0)
expect(fetch).toHaveBeenCalledTimes(2)
})
it('keeps test requests separate from the running poller', async () => {
const { driver, calls } = setup()
await driver.start(vi.fn())
await driver.testConnection()
expect(calls.filter((call) => call.method === 'getUpdates')).toHaveLength(1)
})
it('allows bounded private help with an empty whitelist but no ordinary requests', async () => {
const { driver, polls, calls } = setup([])
polls.push(() => ok([update(1, '/whoami'), update(2, '/whoami'), update(3, 'ordinary'), update(4, '/start')]))
const handler = vi.fn()
await driver.start(handler)
await vi.waitFor(() => expect(calls.filter((call) => call.method === 'sendMessage')).toHaveLength(2))
expect(handler).not.toHaveBeenCalled()
expect(calls.find((call) => call.method === 'sendMessage')?.body.text).toBe('7')
expect(calls.filter((call) => call.method === 'sendMessage')[1]?.body.text).toContain('Offline messages')
})
it('requires numeric whitelist IDs', () => {
expect(() => setup(['@somebody'])).toThrow('numeric')
})
it('preserves all unknown slash text, Bot scope, event ID and millisecond date', async () => {
const { driver, polls } = setup()
polls.push(() => ok(['/ask', '/execute something', '/exec', '/unknown'].map((text, i) => update(i + 1, text))))
const handler = vi.fn(async (message, ack) => { channelInboundTextSchema.parse(message); ack() })
await driver.start(handler)
await vi.waitFor(() => expect(handler).toHaveBeenCalledTimes(4))
expect(handler.mock.calls.map(([message]) => message.text)).toEqual(['/ask', '/execute something', '/exec', '/unknown'])
expect(handler.mock.calls[0]?.[0]).toEqual({ channel: 'telegram', accountId: '42', eventId: '1', senderId: '7', conversationId: '42:7', conversationType: 'direct', text: '/ask', mentioned: false, receivedAt: 1234000 })
})
it('ignores groups, topics, bots, edits, unauthorized users and media captions', async () => {
const { driver, polls, calls } = setup()
polls.push(() => ok([
update(1, '/whoami', { chat: { id: -7, type: 'group' } }),
update(2, 'topic', { message_thread_id: 1 }),
update(3, 'bot', { from: { id: 7, is_bot: true } }),
{ update_id: 4, edited_message: update(4).message },
update(5, 'unauthorized', { from: { id: 9, is_bot: false } }),
update(6, '/start', { photo: [], caption: 'execute caption' }),
update(7, '', { photo: [], from: { id: 9, is_bot: false } }),
update(8, ' '), update(9, 'valid')
]))
const handler = vi.fn(async (_message, ack) => ack())
await driver.start(handler)
await vi.waitFor(() => expect(handler).toHaveBeenCalledTimes(1))
expect(handler.mock.calls[0]?.[0].text).toBe('valid')
expect(calls.filter((call) => call.method === 'sendMessage').map((call) => call.body.text)).toEqual(['Only text messages are supported. Please describe your request in text.'])
})
it('does not advance past unacknowledged input and skips duplicates after acknowledgment', async () => {
const { driver, polls, calls } = setup()
polls.push(() => ok([update(10)]), () => ok([update(10), update(11)]), () => ok([update(11)]))
let attempts = 0
const handler = vi.fn(async (_message, ack) => { if (++attempts > 1) ack() })
await driver.start(handler)
await vi.waitFor(() => expect(calls.filter((call) => call.method === 'getUpdates')).toHaveLength(4), { timeout: 3000 })
expect(calls.filter((call) => call.method === 'getUpdates').map((call) => call.body.offset)).toEqual([0, 0, 12, 12])
expect(handler).toHaveBeenCalledTimes(3)
})
it('stop aborts long polling and waits for fetch cleanup', async () => {
const { driver, calls } = setup()
await driver.start(vi.fn())
await vi.waitFor(() => expect(calls.some((call) => call.method === 'getUpdates')).toBe(true))
await driver.stop()
expect(calls.find((call) => call.method === 'getUpdates')?.signal.aborted).toBe(true)
await expect(driver.send(result(), new AbortController().signal)).rejects.toMatchObject({ name: 'AbortError' })
})
it('cancels validation without starting a poller', async () => {
const { driver, fetch } = setup()
fetch.mockImplementationOnce((_input, init) => pending(init!.signal!))
await driver.start(vi.fn())
const sending = driver.send(result(), new AbortController().signal)
const rejected = expect(sending).rejects.toMatchObject({ name: 'AbortError' })
await driver.stop()
await rejected
expect(fetch).toHaveBeenCalledTimes(1)
})
it('waits on 429 and cancellation interrupts retry_after', async () => {
const { driver, polls, calls, onStatus } = setup()
polls.push(() => error(429, 60))
await driver.start(vi.fn())
await vi.waitFor(() => expect(onStatus).toHaveBeenCalledWith(expect.objectContaining({ state: 'starting', lastError: expect.stringContaining('rate limit') })))
await driver.stop()
expect(calls.filter((call) => call.method === 'getUpdates')).toHaveLength(1)
})
it('backs off transient failures and recovers status', async () => {
const { driver, polls, onStatus, calls } = setup()
polls.push(() => error(503), () => ok([]))
await driver.start(vi.fn())
await vi.waitFor(() => expect(calls.filter((call) => call.method === 'getUpdates').length).toBeGreaterThanOrEqual(2), { timeout: 2500 })
expect(onStatus.mock.calls.some(([status]) => status.lastError?.includes('503'))).toBe(true)
expect(onStatus.mock.calls.at(-1)?.[0]).toEqual({ state: 'running' })
})
it.each([401, 403, 409])('stops polling with a clear redacted %i error', async (code) => {
const { driver, polls, calls, onStatus } = setup()
polls.push(() => error(code))
await driver.start(vi.fn())
await vi.waitFor(() => expect(onStatus).toHaveBeenCalledWith(expect.objectContaining({ state: 'error' })))
const status = onStatus.mock.calls.at(-1)![0]
expect(status.lastError).not.toContain(secret)
expect(status.lastError).not.toContain('https://')
expect(calls.filter((call) => call.method === 'getUpdates')).toHaveLength(1)
})
it('delivers accepted work after a polling conflict without terminating its outbox entry', async () => {
const { driver, polls, calls, onStatus } = setup()
const outbox = new MemoryOutbox()
let finish!: (value: { status: string; output: string }) => void
const executor = vi.fn(() => new Promise<{ status: string; output: string }>((resolve) => { finish = resolve }))
const service = new ChannelService(driver, executor, { allowedSenderIds: ['7'], outbox })
polls.push(() => ok([update(1)]), () => error(409))
try {
await service.start()
await vi.waitFor(() => expect(executor).toHaveBeenCalledOnce())
await vi.waitFor(() => expect(onStatus.mock.calls.at(-1)?.[0].state).toBe('error'))
finish({ status: 'completed', output: 'accepted work result' })
await vi.waitFor(() => expect(calls.filter((call) => call.method === 'sendMessage')).toHaveLength(1))
expect(calls.find((call) => call.method === 'sendMessage')?.body.text).toBe('accepted work result')
expect(outbox.listUndelivered()).toEqual([])
expect(onStatus.mock.calls.at(-1)?.[0].state).toBe('error')
} finally {
await service.stop()
}
})
it('splits pure text at 4096 UTF16 units without splitting emoji and includes error and attachment notice', async () => {
const { driver, calls } = setup()
await driver.start(vi.fn())
const output = `${'a'.repeat(4095)}\uD83D\uDE00${'b'.repeat(4100)}`
await driver.send(result({ output, error: 'task failed', attachments: [{ name: 'file.txt', mimeType: 'text/plain', size: 1, kind: 'file', dataBase64: 'YQ==' }] }), new AbortController().signal)
const parts = calls.filter((call) => call.method === 'sendMessage').map((call) => call.body.text as string)
expect(parts).toHaveLength(3)
expect(parts.every((part) => part.length <= 4096 && !/[\uD800-\uDBFF]$/u.test(part) && !/^[\uDC00-\uDFFF]/u.test(part))).toBe(true)
expect(parts.join('')).toBe(`${output}\n\ntask failed\n\nAttachments are available in GoodBuddy on the desktop; no attachments were sent to Telegram.`)
expect(calls.filter((call) => call.method === 'sendMessage').every((call) => call.body.parse_mode === undefined)).toBe(true)
})
it('resumes unsent chunks within the bounded in-memory progress cache', async () => {
const { driver, sends, calls } = setup()
await driver.start(vi.fn())
sends.push(() => ok(true), () => error(503))
const message = result({ output: 'a'.repeat(4096) + 'tail' })
await expect(driver.send(message, new AbortController().signal)).rejects.toThrow('503')
await driver.send(message, new AbortController().signal)
expect(calls.filter((call) => call.method === 'sendMessage').map((call) => call.body.text)).toEqual(['a'.repeat(4096), 'tail', 'tail'])
})
it('refuses persisted outbox replies from another Bot and unscoped IDs', async () => {
const { driver, calls } = setup()
await driver.start(vi.fn())
for (const conversationId of ['41:7', '7', '42:-7', '42:8']) {
await expect(driver.send(result({ conversationId }), new AbortController().signal)).rejects.toMatchObject({ message: expect.stringContaining('another Bot'), permanent: true, terminal: true })
}
expect(calls.filter((call) => call.method === 'sendMessage')).toHaveLength(0)
})
it('redacts fetch errors, raw Telegram URLs, Token and runtime errors', async () => {
const { driver, fetch, calls } = setup()
fetch.mockRejectedValueOnce(new Error(`https://api.telegram.org/bot${secret}/getMe`))
await expect(driver.testConnection()).rejects.toThrow('network')
await driver.start(vi.fn())
await driver.send(result({ output: secret, error: `https://api.telegram.org/bot${secret}/getUpdates token=private` }), new AbortController().signal)
const text = String(calls.find((call) => call.method === 'sendMessage')?.body.text)
expect(text).not.toContain(secret)
expect(text).not.toContain('api.telegram.org')
expect(text).not.toContain('private')
})
it('cancels outbound rate-limit waits and avoids repeating permanent send failures', async () => {
const { driver, sends, calls } = setup()
await driver.start(vi.fn())
sends.push(() => error(429, 60))
const controller = new AbortController()
const sending = driver.send(result(), controller.signal)
const rejected = expect(sending).rejects.toMatchObject({ name: 'AbortError' })
await vi.waitFor(() => expect(calls.filter((call) => call.method === 'sendMessage')).toHaveLength(1))
controller.abort(new Error(secret))
await rejected
sends.push(() => error(403))
await expect(driver.send(result(), new AbortController().signal)).rejects.toMatchObject({ permanent: true, terminal: true })
await expect(driver.send(result(), new AbortController().signal)).rejects.toThrow('forbidden')
expect(calls.filter((call) => call.method === 'sendMessage')).toHaveLength(2)
})
it('uses Electron net.fetch for the production network path', async () => {
netFetch.mockResolvedValueOnce(ok({ id: 42, is_bot: true })).mockResolvedValueOnce(ok({ url: '' }))
const driver = new TelegramChannelDriver({ secret, allowedSenderIds: [] })
drivers.push(driver)
expect(netFetch).not.toHaveBeenCalled()
await expect(driver.testConnection()).resolves.toEqual({ botId: '42' })
expect(netFetch).toHaveBeenCalledTimes(2)
})
it('honors retry_after before retrying an outbound chunk', async () => {
const { driver, sends, calls } = setup()
await driver.start(vi.fn())
sends.push(() => error(429, 1), () => ok(true))
const started = Date.now()
await driver.send(result(), new AbortController().signal)
expect(Date.now() - started).toBeGreaterThanOrEqual(990)
expect(calls.filter((call) => call.method === 'sendMessage')).toHaveLength(2)
})
it('stop cancels and joins active outbound requests', async () => {
const { driver, fetch, calls } = setup()
await driver.start(vi.fn())
await vi.waitFor(() => expect(calls.some((call) => call.method === 'getUpdates')).toBe(true))
fetch.mockImplementationOnce((_input, init) => pending(init!.signal!))
const sending = driver.send(result(), new AbortController().signal)
const rejected = expect(sending).rejects.toMatchObject({ name: 'AbortError' })
await vi.waitFor(() => expect(fetch).toHaveBeenCalledTimes(4))
await driver.stop()
await rejected
})
it('retains HTTP authentication errors even for non-JSON responses', async () => {
const { driver, fetch } = setup()
fetch.mockResolvedValueOnce(new Response('not JSON', { status: 401 }))
await expect(driver.testConnection()).rejects.toThrow('Token is invalid')
})
it('returns from start while offline, reconnects and releases queued sends only after validation', async () => {
const { driver, fetch, calls, onStatus } = setup()
fetch.mockRejectedValueOnce(new Error(secret))
await driver.start(vi.fn())
const sending = driver.send(result(), new AbortController().signal)
await vi.waitFor(() => expect(onStatus).toHaveBeenCalledWith(expect.objectContaining({ state: 'starting', lastError: expect.stringContaining('Reconnecting') })))
expect(calls).toHaveLength(0)
await sending
expect(calls.map((call) => call.method)).toEqual(['getMe', 'getWebhookInfo', 'getUpdates', 'sendMessage'])
expect(onStatus.mock.calls.at(-1)?.[0]).toEqual({ state: 'running' })
})
it('stop cancels startup backoff and all identity waiters', async () => {
const { driver, fetch, onStatus } = setup()
fetch.mockRejectedValueOnce(new Error(secret))
await driver.start(vi.fn())
const sending = driver.send(result(), new AbortController().signal)
const rejected = expect(sending).rejects.toMatchObject({ name: 'AbortError' })
await vi.waitFor(() => expect(onStatus).toHaveBeenCalledWith(expect.objectContaining({ lastError: expect.stringContaining('Reconnecting') })))
await driver.stop()
await rejected
expect(fetch).toHaveBeenCalledTimes(1)
expect(onStatus.mock.calls.at(-1)?.[0]).toEqual({ state: 'stopped' })
})
it('caller cancellation releases an identity waiter without cancelling startup', async () => {
const { driver, fetch, onStatus } = setup()
let validationSignal: AbortSignal | undefined
fetch.mockImplementationOnce((_input, init) => {
validationSignal = init!.signal!
return pending(validationSignal)
})
await driver.start(vi.fn())
const controller = new AbortController()
const sending = driver.send(result(), controller.signal)
const rejected = expect(sending).rejects.toMatchObject({ name: 'AbortError' })
controller.abort()
await rejected
expect(validationSignal?.aborted).toBe(false)
expect(onStatus.mock.calls.at(-1)?.[0]).toEqual({ state: 'starting' })
})
it.each([401, 409])('reports terminal startup %i and rejects waiting sends without retrying', async (code) => {
const { driver, fetch, onStatus } = setup()
fetch.mockResolvedValueOnce(error(code))
await driver.start(vi.fn())
await expect(driver.send(result(), new AbortController().signal)).rejects.toMatchObject({ code, permanent: true, terminal: true })
expect(fetch).toHaveBeenCalledTimes(1)
expect(onStatus.mock.calls.at(-1)?.[0].state).toBe('error')
})
it('updates the numeric whitelist atomically for subsequent input without restarting', async () => {
const { driver, polls, calls } = setup()
expect(() => driver.updateAllowedSenderIds(['invalid'])).toThrow('numeric')
polls.push(() => ok([update(1), update(2), update(3, 'new user', { from: { id: 8, is_bot: false }, chat: { id: 8, type: 'private' } })]))
const handler = vi.fn(async (_message, ack) => {
ack()
driver.updateAllowedSenderIds(['8'])
})
await driver.start(handler)
await vi.waitFor(() => expect(handler).toHaveBeenCalledTimes(2))
expect(handler.mock.calls.map(([message]) => message.senderId)).toEqual(['7', '8'])
expect(calls.filter((call) => call.method === 'getMe')).toHaveLength(1)
expect(calls.every((call) => !call.signal.aborted)).toBe(true)
// Previously authorized work can still deliver after its sender is removed.
await driver.send(result(), new AbortController().signal)
expect(calls.at(-1)?.method).toBe('sendMessage')
})
it('help rate limits never hold up subsequent inbound requests', async () => {
const { driver, polls, sends, calls } = setup()
sends.push(() => error(429, 86400))
polls.push(() => ok([update(1, '/start'), update(2, 'business request')]))
const handler = vi.fn(async (_message, ack) => ack())
await driver.start(handler)
await vi.waitFor(() => expect(handler).toHaveBeenCalledTimes(1))
expect(calls.filter((call) => call.method === 'sendMessage')).toHaveLength(1)
})
it('testConnection fails fast on rate limiting and cannot replace the running identity', async () => {
const { driver, fetch, calls } = setup()
fetch.mockResolvedValueOnce(error(429, 86400))
await expect(driver.testConnection()).rejects.toMatchObject({ code: 429 })
expect(fetch).toHaveBeenCalledTimes(1)
await driver.start(vi.fn())
await vi.waitFor(() => expect(calls.some((call) => call.method === 'getUpdates')).toBe(true))
fetch.mockResolvedValueOnce(ok({ id: 99, is_bot: true })).mockResolvedValueOnce(ok({ url: '' }))
await expect(driver.testConnection()).resolves.toEqual({ botId: '99' })
await driver.send(result(), new AbortController().signal)
await expect(driver.send(result({ conversationId: '99:7' }), new AbortController().signal)).rejects.toMatchObject({ terminal: true })
})
})
@@ -0,0 +1,347 @@
import { setTimeout as delay } from 'node:timers/promises'
import type { ChannelInboundText, ChannelResultMessage } from '../../shared/channel-contracts'
import type { ChannelRuntimeStatus } from '../../shared/channel-settings-contracts'
import type { ChannelDriver, ChannelInboundHandler } from './channel-driver'
import { redactChannelError } from './channel-service'
export type TelegramChannelDriverOptions = {
secret: string
allowedSenderIds: readonly string[]
fetch?: typeof globalThis.fetch
onStatus?: (status: ChannelRuntimeStatus) => void
}
type BotIdentity = { botId: string; botUsername?: string }
type TelegramUpdate = {
update_id: number
message?: {
message_id: number
date: number
from?: { id: number; is_bot?: boolean }
chat: { id: number; type: string }
text?: string
message_thread_id?: number
is_topic_message?: boolean
[key: string]: unknown
}
}
export class TelegramError extends Error {
constructor(readonly code: number, readonly retryAfter = 1) {
super(code === 401 ? 'Telegram Token is invalid. Replace the Bot Token.'
: code === 409 ? 'Telegram polling conflict. Stop the other poller or remove the existing webhook.'
: code === 403 ? 'Telegram access forbidden. The Bot may be blocked or disabled.'
: code === 429 ? 'Telegram rate limit. Waiting before retrying.'
: code === 0 ? 'Telegram network request failed. Check the network and OS proxy.'
: `Telegram request failed (HTTP ${code}).`)
}
get permanent(): boolean {
return this.code >= 400 && this.code < 500 && this.code !== 429
}
get terminal(): boolean {
return this.permanent
}
}
function cancelled(): Error {
const error = new Error('Telegram operation cancelled.')
error.name = 'AbortError'
return error
}
function checkSignal(signal: AbortSignal): void {
if (signal.aborted) throw cancelled()
}
async function wait(ms: number, signal: AbortSignal): Promise<void> {
try {
await delay(ms, undefined, { signal })
} catch {
throw cancelled()
}
}
function positiveId(value: unknown): value is number {
return typeof value === 'number' && Number.isSafeInteger(value) && value > 0
}
export class TelegramChannelDriver implements ChannelDriver {
readonly channel = 'telegram'
private allowed: ReadonlySet<string> = new Set()
private readonly fetchImpl: typeof globalThis.fetch
private lifecycle = new AbortController()
private readonly operations = new Set<Promise<unknown>>()
private readonly feedback = new Map<string, number>()
private readonly progress = new Map<string, { offset: number; terminal?: TelegramError }>()
private identity?: BotIdentity
private terminalFailure?: TelegramError
private readonly readinessWaiters = new Set<() => void>()
private offset = 0
private pollPromise?: Promise<void>
private stopPromise?: Promise<void>
private lastStatus?: string
constructor(private readonly options: TelegramChannelDriverOptions) {
if (!options.secret.trim() || /[\s/?#]/u.test(options.secret)) {
throw new Error('Telegram Bot Token is invalid.')
}
this.updateAllowedSenderIds(options.allowedSenderIds)
this.fetchImpl = options.fetch ?? (async (input, init) => {
const { net } = await import('electron')
return net.fetch(input instanceof Request ? input.url : String(input), init)
})
}
testConnection(): Promise<BotIdentity> {
return this.track(this.validate(this.lifecycle.signal))
}
updateAllowedSenderIds(ids: readonly string[]): void {
if (ids.some((id) => !/^[1-9]\d*$/u.test(id))) {
throw new Error('Telegram whitelist must contain numeric user IDs.')
}
this.allowed = new Set(ids)
}
start(handler: ChannelInboundHandler): Promise<void> {
if (this.stopPromise) return Promise.reject(new Error('Telegram driver has stopped. Create a new driver to restart.'))
if (this.pollPromise) return Promise.resolve()
this.status({ state: 'starting' })
this.pollPromise = this.poll(handler, this.lifecycle.signal)
return Promise.resolve()
}
stop(): Promise<void> {
if (this.stopPromise) return this.stopPromise
this.lifecycle.abort()
this.stopPromise = (async () => {
await Promise.allSettled([...this.operations, this.pollPromise])
this.feedback.clear()
this.progress.clear()
this.status({ state: 'stopped' })
})()
return this.stopPromise
}
send(message: ChannelResultMessage, signal: AbortSignal): Promise<void> {
return this.track(this.sendResult(message, AbortSignal.any([signal, this.lifecycle.signal])))
}
private async sendResult(message: ChannelResultMessage, signal: AbortSignal): Promise<void> {
checkSignal(signal)
await this.waitForIdentity(signal)
const match = /^([1-9]\d*):([1-9]\d*)$/u.exec(message.conversationId)
if (!this.identity || message.channel !== this.channel || !match ||
match[1] !== this.identity.botId || match[2] !== message.recipientId) {
throw Object.assign(new Error('Telegram reply belongs to another Bot or has an invalid private conversation.'), { permanent: true, terminal: true })
}
const text = this.redact([
message.output?.trim(), message.error?.trim(),
message.attachments?.length ? 'Attachments are available in GoodBuddy on the desktop; no attachments were sent to Telegram.' : undefined
].filter(Boolean).join('\n\n') || 'Request completed.')
const key = `${message.conversationId}:${message.eventId}:${message.status}`
let record = this.progress.get(key)
if (!record) {
record = { offset: 0 }
this.progress.set(key, record)
if (this.progress.size > 128) this.progress.delete(this.progress.keys().next().value!)
}
if (record.terminal) throw record.terminal
while (record.offset < text.length) {
let end = Math.min(record.offset + 4096, text.length)
if (end < text.length && /[\uD800-\uDBFF]/u.test(text.charAt(end - 1)) && /[\uDC00-\uDFFF]/u.test(text.charAt(end))) end -= 1
try {
await this.request('sendMessage', { chat_id: match[2], text: text.slice(record.offset, end) }, signal, true)
record.offset = end
} catch (error) {
if (error instanceof TelegramError && error.permanent) record.terminal = error
throw error
}
}
this.progress.delete(key)
}
private async waitForIdentity(signal: AbortSignal): Promise<void> {
checkSignal(signal)
if (!this.identity && !this.terminalFailure) {
await new Promise<void>((resolve, reject) => {
const finish = () => {
this.readinessWaiters.delete(finish)
signal.removeEventListener('abort', finish)
if (signal.aborted) reject(cancelled())
else resolve()
}
this.readinessWaiters.add(finish)
signal.addEventListener('abort', finish, { once: true })
})
}
checkSignal(signal)
// A polling conflict does not prevent an identified Bot from sending replies.
if (this.terminalFailure && !(this.identity && this.terminalFailure.code === 409)) {
throw this.terminalFailure
}
}
private async validate(signal: AbortSignal): Promise<BotIdentity> {
const me = await this.request<{ id: number; is_bot: boolean; username?: string }>('getMe', {}, signal)
if (!positiveId(me?.id) || me.is_bot !== true) throw new Error('Telegram returned an invalid Bot identity.')
const webhook = await this.request<{ url: string }>('getWebhookInfo', {}, signal)
if (typeof webhook?.url !== 'string') throw new Error('Telegram returned invalid webhook information.')
if (webhook.url) throw new TelegramError(409)
return { botId: String(me.id), ...(me.username ? { botUsername: me.username } : {}) }
}
private async poll(handler: ChannelInboundHandler, signal: AbortSignal): Promise<void> {
let failures = 0
while (!signal.aborted) {
const started = Date.now()
try {
if (!this.identity) {
this.identity = await this.validate(signal)
checkSignal(signal)
for (const notify of this.readinessWaiters) notify()
failures = 0
this.status({ state: 'running' })
}
const updates = await this.request<TelegramUpdate[]>('getUpdates', {
offset: this.offset, timeout: 30, limit: 100, allowed_updates: ['message']
}, signal)
if (!Array.isArray(updates)) throw new TelegramError(0)
for (const update of updates) {
checkSignal(signal)
if (!Number.isSafeInteger(update.update_id) || update.update_id < this.offset) continue
let acknowledged = false
await this.receive(update, handler, () => { acknowledged = true }, signal)
checkSignal(signal)
if (!acknowledged) throw new Error('Telegram inbound message was not acknowledged.')
this.offset = update.update_id + 1
}
failures = 0
this.status({ state: 'running' })
// A server/proxy returning immediately must not turn long polling into a busy loop.
await wait(Math.max(0, 250 - (Date.now() - started)), signal)
} catch (error) {
if (signal.aborted) break
const safe = this.safeError(error)
if (safe instanceof TelegramError && safe.permanent) {
this.terminalFailure = safe
for (const notify of this.readinessWaiters) notify()
this.status({ state: 'error', lastError: safe.message })
break
}
this.status({ state: 'starting', lastError: `${safe.message} Reconnecting.` })
const pause = safe instanceof TelegramError && safe.code === 429
? safe.retryAfter * 1000 : Math.min(30_000, 1000 * 2 ** Math.min(failures++, 5))
try { await wait(pause, signal) } catch { break }
}
}
}
private async receive(update: TelegramUpdate, handler: ChannelInboundHandler, acknowledge: () => void, signal: AbortSignal): Promise<void> {
const message = update.message
if (!message || message.chat?.type !== 'private' || !positiveId(message.chat.id) ||
!positiveId(message.from?.id) || message.from.is_bot !== false ||
message.message_thread_id !== undefined || message.is_topic_message) {
acknowledge()
return
}
const senderId = String(message.from.id)
const media = ['caption', 'photo', 'video', 'animation', 'audio', 'voice', 'document', 'sticker', 'video_note', 'contact', 'location', 'venue', 'poll', 'dice', 'paid_media'].some((key) => message[key] !== undefined)
const text = typeof message.text === 'string' && !media ? message.text.trim() : ''
const command = /^\/(start|whoami)(?:@([^\s]+))?(?:\s|$)/u.exec(text)
const help = command && (!command[2] || command[2].toLowerCase() === this.identity?.botUsername?.toLowerCase())
if (help || (media && this.allowed.has(senderId))) {
const now = Date.now()
const feedbackKey = `${senderId}:${help ? command[1] : 'media'}`
if (now - (this.feedback.get(feedbackKey) ?? -Infinity) >= 5000) {
this.feedback.delete(feedbackKey)
this.feedback.set(feedbackKey, now)
if (this.feedback.size > 128) this.feedback.delete(this.feedback.keys().next().value!)
const reply = help ? command[1] === 'whoami' ? senderId
: 'GoodBuddy accepts private text requests from authorized users. Send /whoami and add your numeric ID to the desktop whitelist. Offline messages retained by Telegram may run when reconnected. After a crash, check desktop task records and results before resending; tasks do not automatically resume.'
: 'Only text messages are supported. Please describe your request in text.'
try {
await this.request('sendMessage', { chat_id: String(message.chat.id), text: reply }, signal)
} catch (error) {
if (signal.aborted) throw error
this.status({ state: 'running', lastError: this.safeError(error).message })
}
}
acknowledge()
return
}
if (!text || !this.allowed.has(senderId) || !Number.isSafeInteger(message.date) || message.date < 0) {
acknowledge()
return
}
const inbound: ChannelInboundText = {
channel: this.channel, accountId: this.identity!.botId,
eventId: String(update.update_id), senderId,
conversationId: `${this.identity!.botId}:${message.chat.id}`,
conversationType: 'direct', text, mentioned: false, receivedAt: message.date * 1000
}
await handler(inbound, acknowledge)
checkSignal(signal)
// Help commands have separate cooldowns so /start does not suppress /whoami.
const typingKey = `${senderId}:typing`
if (Date.now() - (this.feedback.get(typingKey) ?? -Infinity) >= 5000) {
this.feedback.delete(typingKey)
this.feedback.set(typingKey, Date.now())
if (this.feedback.size > 128) this.feedback.delete(this.feedback.keys().next().value!)
try { await this.request('sendChatAction', { chat_id: String(message.chat.id), action: 'typing' }, signal) } catch { /* Feedback must not reject an already queued request. */ }
}
}
private async request<T = unknown>(method: string, body: object, signal: AbortSignal, retryRateLimit = false): Promise<T> {
for (let attempt = 0; ; attempt += 1) {
checkSignal(signal)
const timeout = new AbortController()
const timer = setTimeout(() => timeout.abort(), method === 'getUpdates' ? 40_000 : 15_000)
let failure: TelegramError
try {
const response = await this.fetchImpl(`https://api.telegram.org/bot${this.options.secret}/${method}`, {
method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify(body),
redirect: 'error', signal: AbortSignal.any([signal, timeout.signal])
})
const data = await response.json().catch(() => ({})) as { ok?: boolean; result?: T; error_code?: number; parameters?: { retry_after?: number } }
checkSignal(signal)
if (response.ok && data.ok === true) return data.result as T
const code = response.ok ? (Number.isSafeInteger(data.error_code) ? data.error_code! : 0) : response.status
const seconds = data.parameters?.retry_after
failure = new TelegramError(code, typeof seconds === 'number' && Number.isFinite(seconds) ? Math.max(1, Math.min(seconds, 86400)) : 1)
} catch (error) {
if (signal.aborted) throw cancelled()
failure = error instanceof TelegramError ? error : new TelegramError(0)
} finally {
clearTimeout(timer)
}
if (failure.code !== 429 || !retryRateLimit || attempt >= 3) throw failure
await wait(failure.retryAfter * 1000, signal)
}
}
private redact(value: string): string {
return redactChannelError(value
.replace(/https?:\/\/api\.telegram\.org\/[^\s"'<>]+/giu, '[Telegram URL hidden]')
.split(this.options.secret).join('[Token hidden]'))
}
private safeError(error: unknown): Error {
return error instanceof TelegramError ? error : new Error('Telegram operation failed. Check the connection and inbound queue.')
}
private status(status: ChannelRuntimeStatus): void {
const key = JSON.stringify(status)
if (key === this.lastStatus) return
this.lastStatus = key
this.options.onStatus?.(status)
}
private track<T>(promise: Promise<T>): Promise<T> {
this.operations.add(promise)
void promise.finally(() => this.operations.delete(promise)).catch(() => undefined)
return promise
}
}
+134 -2
View File
@@ -35,6 +35,7 @@ import type {
} from '../shared/ssh-host-contracts'
import { defaultKnowledgeOntologySettings } from '../shared/knowledge-ontology'
import { AssistantDatabase } from './assistant/assistant-database'
import { ChannelSettingsStore } from './channels/channel-settings-store'
import { SubagentService, createSubagentRuntime } from './assistant/subagent-service'
import { ConversationAttachmentStorage } from './conversation-attachment-storage'
import { DocumentResultStorage } from './document-result-storage'
@@ -350,6 +351,7 @@ const electronMocks = vi.hoisted(() => {
return {
handlers,
invoke: vi.fn(),
fetch: vi.fn<typeof fetch>(),
handle: vi.fn((channel: string, handler: InvokeHandler) => {
handlers.set(channel, handler)
}),
@@ -1391,6 +1393,7 @@ describe('registerIpcHandlers model download source routing', () => {
})
vi.mock('electron', () => ({
net: { fetch: electronMocks.fetch },
contextBridge: { exposeInMainWorld: (name: string, value: unknown) => vi.stubGlobal(name, value) },
ipcRenderer: { invoke: electronMocks.invoke },
webUtils: {},
@@ -5374,7 +5377,8 @@ describe('registerIpcHandlers agent terminal state', () => {
obsidianService?: ObsidianService,
nativeClientCoordinator?: Parameters<typeof registerIpcHandlers>[44],
nativeTerminalManager?: Parameters<typeof registerIpcHandlers>[40],
applicationSettingsStore?: ApplicationSettingsStore
applicationSettingsStore?: ApplicationSettingsStore,
channelSettingsStore?: ChannelSettingsStore
) {
runtime.getStatus ??= vi.fn(async () => ({
id: runtime.runtimeId ?? 'model', available: true,
@@ -5625,6 +5629,7 @@ describe('registerIpcHandlers agent terminal state', () => {
args[44] = nativeClientCoordinator
args[40] = nativeTerminalManager
if (applicationSettingsStore) args[14] = applicationSettingsStore
args[13] = channelSettingsStore
runtimeFactoryMocks.createModelProfileRuntime.mockReturnValue(runtime)
const dispose = registerIpcHandlers(...args)
return {
@@ -12150,6 +12155,131 @@ describe('registerIpcHandlers agent terminal state', () => {
}
})
it('routes Telegram settings and incoming messages through real channel handlers without modes', async () => {
const root = await mkdtemp(join(tmpdir(), 'goodbuddy-ipc-telegram-'))
const database = new AssistantDatabase(':memory:')
database.initialize(root)
database.ensureChannelProjects(root)
const routeConversation = vi.spyOn(database, 'getOrCreateRemoteConversation')
const settingsStore = new ChannelSettingsStore(join(root, 'channels.json'), {
isAvailable: () => true,
encrypt: value => Buffer.from(value),
decrypt: value => value.toString()
}, {})
const requests: AgentExecutionRequest[] = []
const runtime = {
capability: 'chat', supportsToolExecution: true,
async *run(request: AgentExecutionRequest, _signal: AbortSignal, authorize?: unknown) {
expect(authorize).toBeUndefined()
requests.push(request)
yield { requestId: request.requestId, type: 'text', delta: 'Telegram result' }
yield { requestId: request.requestId, type: 'done' }
}
}
let delivered = false
const replies: unknown[] = []
electronMocks.fetch.mockImplementation(async (input, init) => {
const method = String(input).split('/').at(-1)
let result: unknown
if (method === 'getMe') result = { id: 456, is_bot: true, username: 'ipc_test_bot' }
else if (method === 'getWebhookInfo') result = { url: '' }
else if (method === 'getUpdates') {
result = delivered ? [] : [{
update_id: 789,
message: {
message_id: 1, date: Math.floor(Date.now() / 1000),
from: { id: 123, is_bot: false }, chat: { id: 123, type: 'private' },
text: '/execute preserve this literal prefix'
}
}]
delivered = true
} else if (method === 'sendMessage') {
replies.push(JSON.parse(String(init?.body)))
result = { message_id: 2 }
} else if (method === 'sendChatAction') result = true
else throw new Error(`Unexpected Telegram method: ${method}`)
return new Response(JSON.stringify({ ok: true, result }))
})
let harness: ReturnType<typeof createHarness> | undefined
try {
harness = createHarness(runtime, undefined, undefined, false, undefined, undefined,
undefined, false, undefined, undefined, undefined, undefined, database,
undefined, undefined, undefined, undefined, undefined, settingsStore)
const event = trustedEvent(harness.webContents)
const get = electronMocks.handlers.get(ipcChannels.channelSettingsGet)!
const apply = electronMocks.handlers.get(ipcChannels.channelSettingsApply)!
const test = electronMocks.handlers.get(ipcChannels.channelSettingsTest)!
const settings = {
enabled: false, secret: { action: 'replace', value: '456:test-token' },
allowedSenderIds: ['123'], allowGroupMessages: false
}
await expect(test(event, { channel: 'telegram', settings })).resolves.toEqual({
channel: 'telegram', ok: true, botUsername: 'ipc_test_bot'
})
await expect(get(event)).resolves.toMatchObject({ telegram: { secretConfigured: false } })
expect(delivered).toBe(false)
expect(() => test(event, {
channel: 'telegram', settings: { ...settings, allowedSenderIds: ['username'] }
})).toThrow()
expect(() => apply(event, { telegram: { ...settings, allowGroupMessages: true } })).toThrow()
expect(() => test({ sender: {}, senderFrame: harness!.webContents.mainFrame }, {
channel: 'telegram', settings
})).toThrow('拒绝来自未知窗口的 IPC 请求')
await expect(apply(event, { telegram: settings })).resolves.toMatchObject({
telegram: { enabled: false, secretConfigured: true, allowedSenderIds: ['123'] }
})
await expect(test(event, { channel: 'telegram' })).resolves.toEqual({
channel: 'telegram', ok: true, botUsername: 'ipc_test_bot'
})
await apply(event, { telegram: { ...settings, enabled: true, secret: { action: 'keep' } } })
await vi.waitFor(() => expect(replies).toContainEqual({ chat_id: '123', text: 'Telegram result' }))
expect(requests).toHaveLength(1)
const request = requests[0]!
const project = database.listProjects().find(item => item.channel === 'telegram')!
expect(project).toMatchObject({ kind: 'channel', name: 'Telegram' })
expect(routeConversation).toHaveBeenCalledExactlyOnceWith(expect.objectContaining({
projectId: project.id, channel: 'telegram', accountId: '456',
externalConversationId: '456:123', conversationType: 'direct'
}))
expect(database.getConversation(request.conversationId!)).toMatchObject({
projectId: project.id,
remote: { channel: 'telegram', conversationType: 'direct' },
messages: [
expect.objectContaining({ role: 'user', content: '/execute preserve this literal prefix' }),
expect.objectContaining({ role: 'assistant', content: 'Telegram result' })
]
})
expect(database.getTask(request.requestId)).toMatchObject({
projectId: project.id, conversationId: request.conversationId,
origin: 'delegation', status: 'completed', instructions: '/execute preserve this literal prefix'
})
expect(request.prompt).toContain('/execute preserve this literal prefix')
expect(request).not.toHaveProperty('workMode')
expect(request).not.toHaveProperty('mode')
expect(harness.webContents.send).toHaveBeenCalledWith(ipcChannels.channelStatusChanged, {
channel: 'telegram', status: { state: 'starting' }
})
expect(harness.webContents.send).toHaveBeenCalledWith(ipcChannels.channelStatusChanged, {
channel: 'telegram', status: { state: 'running' }
})
const history = database.getConversation(request.conversationId!)
await expect(apply(event, { telegram: { ...settings, secret: { action: 'clear' } } })).resolves.toMatchObject({
telegram: { enabled: false, secretConfigured: false, status: { state: 'disabled' } }
})
expect(harness.webContents.send).toHaveBeenCalledWith(ipcChannels.channelStatusChanged, {
channel: 'telegram', status: { state: 'disabled' }
})
expect((await settingsStore.resolve('telegram')).secret).toBeUndefined()
expect(database.getProject(project.id)).toEqual(project)
expect(database.getConversation(request.conversationId!)).toEqual(history)
} finally {
await harness?.dispose()
electronMocks.fetch.mockReset()
database.close()
await rm(root, { recursive: true, force: true })
}
})
it('bridges explanation-only channel requests without an authorizer', async () => {
let received:
| {
@@ -12231,7 +12361,8 @@ describe('registerIpcHandlers agent terminal state', () => {
it.each([
['weixin', '微信 ClawBot'],
['wecom', '企业微信'],
['dingtalk', '钉钉']
['dingtalk', '钉钉'],
['telegram', 'Telegram']
] as const)(
'grants enabled Magic Notes write tools to %s channel requests',
async (channel, channelLabel) => {
@@ -12341,6 +12472,7 @@ describe('registerIpcHandlers agent terminal state', () => {
)
})
expect(receivedRequest?.prompt).toContain('读取我的笔记')
expect(receivedRequest).not.toHaveProperty('workMode')
expect(knowledgeGateway.revoke).toHaveBeenCalledWith(
'channel-notes-capability'
)
+15
View File
@@ -81,6 +81,7 @@ import {
import {
channelSettingsApplySchema,
dingTalkChannelSettingsInputSchema,
telegramChannelSettingsInputSchema,
weComChannelSettingsInputSchema
} from '../shared/channel-settings-contracts'
import { applicationSettingsUpdateSchema, defaultSupervisorModelConcurrency } from '../shared/application-settings-contracts'
@@ -346,6 +347,12 @@ const channelSettingsTestRequestSchema = z.discriminatedUnion('channel', [
channel: z.literal('dingtalk'),
settings: dingTalkChannelSettingsInputSchema.optional()
})
.strict(),
z
.object({
channel: z.literal('telegram'),
settings: telegramChannelSettingsInputSchema.optional()
})
.strict()
])
@@ -3057,6 +3064,11 @@ export function registerIpcHandlers(
const channelManager = channelSettingsStore
? new ChannelManager(channelSettingsStore, channelExecutor, {
launchWechatSidecar,
onStatusChanged: (change) => {
if (!window.isDestroyed()) {
window.webContents.send(ipcChannels.channelStatusChanged, change)
}
},
dedupStore: new SqliteChannelDedupStore(assistantDatabase),
outbox: new SqliteChannelOutbox(assistantDatabase)
})
@@ -5078,6 +5090,9 @@ export function registerIpcHandlers(
throw new Error('消息通道设置服务不可用')
}
const request = channelSettingsTestRequestSchema.parse(input)
if (request.channel === 'telegram') {
return channelManager.testConnection('telegram', request.settings)
}
return request.channel === 'wecom'
? channelManager.testConnection('wecom', request.settings)
: channelManager.testConnection('dingtalk', request.settings)