diff --git a/backend/__tests__/unit/routes/discord.management-auth.test.js b/backend/__tests__/unit/routes/discord.management-auth.test.js index 40583e034..74fce564e 100644 --- a/backend/__tests__/unit/routes/discord.management-auth.test.js +++ b/backend/__tests__/unit/routes/discord.management-auth.test.js @@ -180,4 +180,19 @@ describe('discord management route auth', () => { expect(res.status).toBe(403); expect(DiscordIntegration.findOneAndDelete).not.toHaveBeenCalled(); }); + + it('scopes legacy uninstall lookup to Discord integrations', async () => { + Integration.findOne.mockResolvedValue(null); + + const res = await request(app) + .delete('/api/discord/uninstall/install-1') + .set('Authorization', 'Bearer user-token'); + + expect(res.status).toBe(404); + expect(Integration.findOne).toHaveBeenCalledWith({ + installationId: 'install-1', + type: 'discord', + }); + expect(Integration.findByIdAndDelete).not.toHaveBeenCalled(); + }); }); diff --git a/backend/__tests__/unit/routes/installables.test.js b/backend/__tests__/unit/routes/installables.test.js new file mode 100644 index 000000000..19cb7bcc5 --- /dev/null +++ b/backend/__tests__/unit/routes/installables.test.js @@ -0,0 +1,189 @@ +const request = require('supertest'); +const express = require('express'); + +jest.mock('../../../middleware/auth', () => (req, res, next) => { + if (!req.header('Authorization')) return res.status(401).json({ error: 'Unauthorized' }); + req.user = { id: '64b64c48c4f37a6b2f34c111' }; + return next(); +}); + +jest.mock('../../../middleware/integrationRateLimit', () => ({ + writeIntegrationsRateLimit: (_req, _res, next) => next(), +})); + +jest.mock('../../../models/Pod', () => ({ findById: jest.fn() })); +jest.mock('../../../services/installable/installableInstallationService', () => ({ + install: jest.fn(), + uninstall: jest.fn(), + InstallLockLostError: class InstallLockLostError extends Error {}, + InstallableAlreadyInstalledError: function InstallableAlreadyInstalledError(boundPodId) { + this.message = 'This connector is already installed for another pod.'; + this.boundPodId = boundPodId; + }, + InstallableNotFoundError: class InstallableNotFoundError extends Error {}, + InstallableProjectionError: class InstallableProjectionError extends Error {}, + InstallInProgressError: function InstallInProgressError(boundPodId) { + this.message = 'This install is still in progress; try again shortly.'; + this.boundPodId = boundPodId; + }, +})); + +const Pod = require('../../../models/Pod'); +const installationService = require('../../../services/installable/installableInstallationService'); +const installableRoutes = require('../../../routes/installables'); + +const app = express(); +app.use(express.json()); +app.use('/api/installables', installableRoutes); + +const auth = { Authorization: 'Bearer test-token' }; +const podId = '64b64c48c4f37a6b2f34c222'; + +describe('installable connector routes', () => { + beforeEach(() => { + jest.clearAllMocks(); + }); + + it('rejects a non-member before any install row is claimed', async () => { + Pod.findById.mockResolvedValue({ + _id: podId, + createdBy: { toString: () => 'someone-else' }, + members: ['someone-else'], + }); + + const res = await request(app) + .post('/api/installables/telegram/install') + .set(auth) + .send({ podId }); + + expect(res.status).toBe(403); + expect(installationService.install).not.toHaveBeenCalled(); + }); + + it('rejects an invalid podId without querying a pod or claiming an install', async () => { + const response = await request(app) + .post('/api/installables/telegram/install') + .set('Authorization', 'Bearer valid') + .send({ podId: 'not-an-object-id' }); + + expect(response.status).toBe(400); + expect(response.body.error).toBe('podId must be a valid ObjectId'); + expect(Pod.findById).not.toHaveBeenCalled(); + expect(installationService.install).not.toHaveBeenCalled(); + }); + + it('derives the install target from auth and accepts only the selected pod', async () => { + Pod.findById.mockResolvedValue({ + _id: podId, + createdBy: { toString: () => 'someone-else' }, + members: ['64b64c48c4f37a6b2f34c111'], + }); + installationService.install.mockResolvedValue({ + httpStatus: 200, + state: 'active', + installation: { _id: 'install-1' }, + integration: { _id: 'integration-1' }, + }); + + const res = await request(app) + .post('/api/installables/telegram/install') + .set(auth) + .send({ podId }); + + expect(res.status).toBe(200); + expect(installationService.install).toHaveBeenCalledWith({ + installableId: 'telegram', + installedBy: '64b64c48c4f37a6b2f34c111', + podId, + }); + }); + + it('returns typed 409 when activation loses its projection', async () => { + Pod.findById.mockResolvedValue({ + _id: podId, + createdBy: { toString: () => 'someone-else' }, + members: ['64b64c48c4f37a6b2f34c111'], + }); + installationService.install.mockRejectedValue( + new installationService.InstallLockLostError('install lock lost'), + ); + + const res = await request(app) + .post('/api/installables/telegram/install') + .set(auth) + .send({ podId }); + + expect(res.status).toBe(409); + expect(res.body.code).toBe('install_lock_lost'); + }); + + it('returns the bound pod when an active connector is requested for another pod', async () => { + Pod.findById.mockResolvedValue({ + _id: podId, + createdBy: { toString: () => 'someone-else' }, + members: ['64b64c48c4f37a6b2f34c111'], + }); + installationService.install.mockRejectedValue( + new installationService.InstallableAlreadyInstalledError('64b64c48c4f37a6b2f34c333'), + ); + + const res = await request(app) + .post('/api/installables/telegram/install') + .set(auth) + .send({ podId }); + + expect(res.status).toBe(409); + expect(res.body).toMatchObject({ + code: 'already_installed', + boundPodId: '64b64c48c4f37a6b2f34c333', + }); + }); + + it('returns the pending pod when a fresh claim targets another pod', async () => { + Pod.findById.mockResolvedValue({ + _id: podId, + createdBy: { toString: () => 'someone-else' }, + members: ['64b64c48c4f37a6b2f34c111'], + }); + installationService.install.mockRejectedValue( + new installationService.InstallInProgressError('64b64c48c4f37a6b2f34c333'), + ); + + const res = await request(app) + .post('/api/installables/telegram/install') + .set(auth) + .send({ podId }); + + expect(res.status).toBe(409); + expect(res.body).toMatchObject({ + code: 'install_in_progress', + boundPodId: '64b64c48c4f37a6b2f34c333', + }); + }); + + it('cannot use an uninstall body to target another user installation', async () => { + installationService.uninstall.mockResolvedValue({ _id: 'install-1', status: 'uninstalled' }); + + const res = await request(app) + .delete('/api/installables/telegram/install') + .set(auth) + .send({ installationId: 'another-users-install' }); + + expect(res.status).toBe(200); + expect(installationService.uninstall).toHaveBeenCalledWith({ + installableId: 'telegram', + installedBy: '64b64c48c4f37a6b2f34c111', + }); + }); + + it('does not report a concurrent revocation as disconnected', async () => { + installationService.uninstall.mockResolvedValue({ _id: 'install-1', status: 'uninstalling' }); + + const res = await request(app) + .delete('/api/installables/telegram/install') + .set(auth); + + expect(res.status).toBe(202); + expect(res.body.status).toBe('uninstalling'); + }); +}); diff --git a/backend/__tests__/unit/services/installableEventHandlers.test.js b/backend/__tests__/unit/services/installableEventHandlers.test.js new file mode 100644 index 000000000..3400b91d1 --- /dev/null +++ b/backend/__tests__/unit/services/installableEventHandlers.test.js @@ -0,0 +1,169 @@ +// @ts-nocheck + +const mongoose = require('mongoose'); + +const Installable = require('../../../models/Installable'); +const InstallableInstallation = require('../../../models/InstallableInstallation'); +const Integration = require('../../../models/Integration'); +const { + install, +} = require('../../../services/installable/installableInstallationService'); +const { + dispatch, + eventHandlers, +} = require('../../../services/installable/eventHandlers'); +const telegramSend = require('../../../services/telegramService'); +const { TELEGRAM_CONNECTOR } = require('../../../scripts/seed-builtin-connectors'); +const { + setupMongoDb, + closeMongoDb, + clearMongoDb, +} = require('../../utils/testUtils'); + +const freshId = () => new mongoose.Types.ObjectId().toString(); + +describe('installable event dispatcher', () => { + let originalTelegramHandler; + + beforeAll(async () => { + await setupMongoDb(); + await InstallableInstallation.syncIndexes(); + }); + + afterAll(async () => { + await closeMongoDb(); + }); + + beforeEach(async () => { + await clearMongoDb(); + await Installable.create({ + ...TELEGRAM_CONNECTOR, + stats: { totalInstalls: 0, activeInstalls: 0, forkCount: 0 }, + }); + originalTelegramHandler = eventHandlers['telegram.relay']; + }); + + afterEach(() => { + eventHandlers['telegram.relay'] = originalTelegramHandler; + }); + + it('selects only the event pod connector before invoking its handler', async () => { + const podA = freshId(); + const podB = freshId(); + await install({ installableId: 'telegram', installedBy: freshId(), podId: podA }); + await install({ installableId: 'telegram', installedBy: freshId(), podId: podB }); + const relay = jest.fn().mockResolvedValue(undefined); + eventHandlers['telegram.relay'] = relay; + + await dispatch('chat.message', { + podId: podA, + agentUsername: 'kai', + displayName: 'Kai', + content: '[ESCALATE] needs review', + podMessageId: 'message-a', + }); + + expect(relay).toHaveBeenCalledTimes(1); + expect(relay).toHaveBeenCalledWith(expect.objectContaining({ + podId: podA, + agentUsername: 'kai', + displayName: 'Kai', + content: '[ESCALATE] needs review', + podMessageId: 'message-a', + integration: expect.objectContaining({ podId: expect.anything() }), + })); + expect(String(relay.mock.calls[0][0].integration.podId)).toBe(podA); + }); + + it('relays each same-pod connector to its own selected Telegram chat', async () => { + const podId = freshId(); + const first = await install({ installableId: 'telegram', installedBy: freshId(), podId }); + const second = await install({ installableId: 'telegram', installedBy: freshId(), podId }); + await Integration.updateOne( + { _id: first.integration._id }, + { $set: { 'config.chatId': 'chat-a', 'config.chatType': 'private' } }, + ); + await Integration.updateOne( + { _id: second.integration._id }, + { $set: { 'config.chatId': 'chat-b', 'config.chatType': 'private' } }, + ); + const previousToken = process.env.TELEGRAM_BOT_TOKEN; + process.env.TELEGRAM_BOT_TOKEN = 'test-bot-token'; + const sendMessage = jest.spyOn(telegramSend, 'sendMessage') + .mockResolvedValueOnce({ messageId: 1 }) + .mockResolvedValueOnce({ messageId: 2 }); + + try { + await dispatch('chat.message', { + podId, + agentUsername: 'kai', + displayName: 'Kai', + content: 'Both subscriptions should receive this', + podMessageId: 'message-fanout', + }); + + expect(sendMessage).toHaveBeenCalledTimes(2); + expect(new Set(sendMessage.mock.calls.map((call) => call[1]))).toEqual( + new Set(['chat-a', 'chat-b']), + ); + } finally { + sendMessage.mockRestore(); + if (previousToken === undefined) delete process.env.TELEGRAM_BOT_TOKEN; + else process.env.TELEGRAM_BOT_TOKEN = previousToken; + } + }); + + it('does not invoke a handler when the event pod has no installation', async () => { + const relay = jest.fn().mockResolvedValue(undefined); + eventHandlers['telegram.relay'] = relay; + + await dispatch('chat.message', { + podId: freshId(), + agentUsername: 'kai', + displayName: 'Kai', + content: 'No connector', + podMessageId: 'message-none', + }); + + expect(relay).not.toHaveBeenCalled(); + }); + + it('continues to dispatch an existing direct Telegram integration', async () => { + const podId = freshId(); + await Integration.create({ + podId, + type: 'telegram', + status: 'pending', + createdBy: freshId(), + isActive: true, + config: { liveRelay: true, chatId: 'chat-1', chatType: 'private' }, + }); + const relay = jest.fn().mockResolvedValue(undefined); + eventHandlers['telegram.relay'] = relay; + + await dispatch('chat.message', { + podId, + agentUsername: 'kai', + displayName: 'Kai', + content: 'Legacy connector still works', + podMessageId: 'message-legacy', + }); + + expect(relay).toHaveBeenCalledTimes(1); + expect(relay.mock.calls[0][0].podId).toBe(podId); + }); + + it('contains an individual handler failure', async () => { + const podId = freshId(); + await install({ installableId: 'telegram', installedBy: freshId(), podId }); + eventHandlers['telegram.relay'] = jest.fn().mockRejectedValue(new Error('provider unavailable')); + + await expect(dispatch('chat.message', { + podId, + agentUsername: 'kai', + displayName: 'Kai', + content: 'Will not fail the post', + podMessageId: 'message-fail', + })).resolves.toBeUndefined(); + }); +}); diff --git a/backend/__tests__/unit/services/installableInstallationService.test.js b/backend/__tests__/unit/services/installableInstallationService.test.js new file mode 100644 index 000000000..b52166703 --- /dev/null +++ b/backend/__tests__/unit/services/installableInstallationService.test.js @@ -0,0 +1,540 @@ +// @ts-nocheck + +const mongoose = require('mongoose'); + +const Installable = require('../../../models/Installable'); +const InstallableInstallation = require('../../../models/InstallableInstallation'); +const Integration = require('../../../models/Integration'); +const { + install, + uninstall, + InstallLockLostError, + InstallableAlreadyInstalledError, + InstallableProjectionError, +} = require('../../../services/installable/installableInstallationService'); +const { sweep } = require('../../../services/installable/installableReconciler'); +const { TELEGRAM_CONNECTOR } = require('../../../scripts/seed-builtin-connectors'); +const { + setupMongoDb, + closeMongoDb, + clearMongoDb, +} = require('../../utils/testUtils'); +const { projectorRegistry } = require('../../../services/installable/projectors'); + +const ids = () => ({ + userId: new mongoose.Types.ObjectId().toString(), + podId: new mongoose.Types.ObjectId().toString(), +}); + +describe('installable connector projection', () => { + beforeAll(async () => { + await setupMongoDb(); + await InstallableInstallation.syncIndexes(); + }); + + afterAll(async () => { + await closeMongoDb(); + }); + + beforeEach(async () => { + await clearMongoDb(); + await Installable.create({ + ...TELEGRAM_CONNECTOR, + stats: { totalInstalls: 0, activeInstalls: 0, forkCount: 0 }, + }); + }); + + it('projects Telegram once, mints last, and shares one Integration between components', async () => { + const { userId, podId } = ids(); + + const result = await install({ installableId: 'telegram', installedBy: userId, podId }); + + expect(result.httpStatus).toBe(200); + expect(result.installation.status).toBe('active'); + expect(result.installation.components).toHaveLength(2); + expect(result.installation.components.every((component) => component.status === 'active')).toBe(true); + + const integration = await Integration.findOne({ installationId: String(result.installation._id) }); + expect(integration).toMatchObject({ + installationId: String(result.installation._id), + type: 'telegram', + isActive: true, + podId: expect.objectContaining({ toString: expect.any(Function) }), + }); + expect(String(integration.podId)).toBe(podId); + expect(String(result.installation.boundPodId)).toBe(podId); + expect(integration.createdBy.toString()).toBe(userId); + expect(integration.config.linkedUserId).toBe(userId); + expect(integration.config.liveRelay).toBe(true); + expect(integration.config.relayAllAgentMessages).toBe(true); + expect(integration.config.connectCode).toMatch(/^[a-f0-9]{32}$/); + expect(integration.config.connectCodeExpiresAt).toBeInstanceOf(Date); + + const projectedIds = result.installation.components.map((component) => ( + String(component.projectionIds.get('integrationId')) + )); + expect(new Set(projectedIds)).toEqual(new Set([String(integration._id)])); + }); + + it('returns the existing active install and only one parent during concurrent installs', async () => { + const { userId, podId } = ids(); + const webhookProjector = projectorRegistry.get('webhook'); + const project = jest.spyOn(webhookProjector, 'project'); + const [first, second] = await Promise.all([ + install({ installableId: 'telegram', installedBy: userId, podId }), + install({ installableId: 'telegram', installedBy: userId, podId }), + ]); + + expect([first.httpStatus, second.httpStatus]).toEqual(expect.arrayContaining([200])); + expect(await InstallableInstallation.countDocuments({ installableId: 'telegram' })).toBe(1); + expect(await Integration.countDocuments({ type: 'telegram' })).toBe(1); + expect(project).toHaveBeenCalledTimes(1); + project.mockRestore(); + + const retry = await install({ installableId: 'telegram', installedBy: userId, podId }); + expect(retry.httpStatus).toBe(200); + expect(String(retry.installation._id)).toBe(String(first.installation._id)); + }); + + it('refuses a different pod instead of reporting an existing install as a success', async () => { + const { userId, podId } = ids(); + const otherPodId = new mongoose.Types.ObjectId().toString(); + await install({ installableId: 'telegram', installedBy: userId, podId }); + const before = await InstallableInstallation.findOne({ installableId: 'telegram' }); + + await expect(install({ installableId: 'telegram', installedBy: userId, podId: otherPodId })) + .rejects.toMatchObject({ + code: 'already_installed', + boundPodId: podId, + }); + await expect(install({ installableId: 'telegram', installedBy: userId, podId: otherPodId })) + .rejects.toBeInstanceOf(InstallableAlreadyInstalledError); + + const integration = await Integration.findOne({ type: 'telegram' }); + expect(String(integration.podId)).toBe(podId); + const after = await InstallableInstallation.findById(before._id); + expect(after).toMatchObject({ + status: 'active', + claimId: before.claimId, + claimedAt: before.claimedAt, + }); + }); + + it('uses an existing active projection as the binding authority after parent completion recovers', async () => { + const { userId, podId } = ids(); + const retryPodId = new mongoose.Types.ObjectId().toString(); + const first = await install({ installableId: 'telegram', installedBy: userId, podId }); + + // A retry can update claim intent before boot reconciliation observes the + // already-active projection and finishes the parent. The projection, not + // that newer intent, decides where the live relay routes. + await InstallableInstallation.updateOne( + { _id: first.installation._id }, + { $set: { boundPodId: new mongoose.Types.ObjectId(retryPodId), status: 'active' } }, + ); + + await expect(install({ installableId: 'telegram', installedBy: userId, podId: retryPodId })) + .rejects.toMatchObject({ + code: 'already_installed', + boundPodId: podId, + }); + const integration = await Integration.findById(first.integration._id); + expect(String(integration.podId)).toBe(podId); + }); + + it('reports a fresh claim in progress without re-targeting it', async () => { + const { userId, podId } = ids(); + const otherPodId = new mongoose.Types.ObjectId().toString(); + const targetId = new mongoose.Types.ObjectId(userId); + const parent = await InstallableInstallation.create({ + installableId: 'telegram', + installableVersion: '1.0.0', + targetType: 'user', + targetId, + scope: 'user', + boundPodId: new mongoose.Types.ObjectId(podId), + installedBy: targetId, + installSource: 'direct', + status: 'installing', + claimId: 'owner-a', + claimedAt: new Date(), + }); + + const waiting = await install({ installableId: 'telegram', installedBy: userId, podId }); + expect(waiting).toMatchObject({ httpStatus: 202, state: 'installing', boundPodId: podId }); + + await expect(install({ installableId: 'telegram', installedBy: userId, podId: otherPodId })) + .rejects.toMatchObject({ code: 'install_in_progress', boundPodId: podId }); + + const after = await InstallableInstallation.findById(parent._id); + expect(after).toMatchObject({ + status: 'installing', + claimId: 'owner-a', + claimedAt: parent.claimedAt, + boundPodId: new mongoose.Types.ObjectId(podId), + }); + expect(await Integration.countDocuments({ installationId: String(parent._id) })).toBe(0); + }); + + it('survives repeated component failures and claims the same projection on retry', async () => { + const { userId, podId } = ids(); + await Installable.updateOne( + { installableId: 'telegram' }, + { $set: { 'components.1.eventHandler': 'internal:missing' } }, + ); + + await expect(install({ installableId: 'telegram', installedBy: userId, podId })) + .rejects.toBeInstanceOf(InstallableProjectionError); + + const failed = await InstallableInstallation.findOne({ installableId: 'telegram' }); + const inactive = await Integration.findOne({ installationId: String(failed._id) }); + expect(failed.status).toBe('error'); + expect(inactive.isActive).toBe(false); + expect(inactive.config.connectCode).toBeUndefined(); + + await expect(install({ installableId: 'telegram', installedBy: userId, podId })) + .rejects.toBeInstanceOf(InstallableProjectionError); + const failedTwice = await InstallableInstallation.findById(failed._id); + expect(failedTwice.status).toBe('error'); + expect(await Integration.countDocuments({ installationId: String(failed._id) })).toBe(1); + + await Installable.updateOne( + { installableId: 'telegram' }, + { $set: { 'components.1.eventHandler': 'internal:telegram.relay' } }, + ); + const retryPodId = new mongoose.Types.ObjectId().toString(); + const retried = await install({ installableId: 'telegram', installedBy: userId, podId: retryPodId }); + expect(String(retried.installation._id)).toBe(String(failed._id)); + expect(retried.integration.config.connectCode).toMatch(/^[a-f0-9]{32}$/); + expect(String(retried.integration.podId)).toBe(retryPodId); + expect(await Integration.countDocuments({ installationId: String(failed._id) })).toBe(1); + }); + + it('returns 202 for a fresh claim and takes over the same parent after its lease expires', async () => { + const { userId, podId } = ids(); + const targetId = new mongoose.Types.ObjectId(userId); + const fresh = await InstallableInstallation.create({ + installableId: 'telegram', + installableVersion: '1.0.0', + targetType: 'user', + targetId, + scope: 'user', + installedBy: targetId, + installSource: 'direct', + status: 'installing', + claimId: 'owner-a', + claimedAt: new Date(), + }); + + const waiting = await install({ installableId: 'telegram', installedBy: userId, podId }); + expect(waiting.httpStatus).toBe(202); + expect(String(waiting.installation._id)).toBe(String(fresh._id)); + + fresh.claimedAt = new Date(Date.now() - 61_000); + await fresh.save(); + const takenOver = await install({ installableId: 'telegram', installedBy: userId, podId }); + expect(takenOver.httpStatus).toBe(200); + expect(String(takenOver.installation._id)).toBe(String(fresh._id)); + expect(takenOver.installation.claimId).not.toBe('owner-a'); + expect(takenOver.integration.config.connectCode).toMatch(/^[a-f0-9]{32}$/); + }); + + it('resumes a stale activating generation without re-projecting or re-minting its code', async () => { + const { userId, podId } = ids(); + const first = await install({ installableId: 'telegram', installedBy: userId, podId }); + const installation = await InstallableInstallation.findById(first.installation._id); + const integration = await Integration.findById(first.integration._id); + const originalCode = integration.config.connectCode; + + installation.status = 'activating'; + installation.claimedAt = new Date(Date.now() - 61_000); + await installation.save(); + + const resumed = await install({ installableId: 'telegram', installedBy: userId, podId }); + expect(resumed.httpStatus).toBe(200); + expect(String(resumed.installation._id)).toBe(String(first.installation._id)); + expect(resumed.integration.config.connectCode).toBe(originalCode); + expect(await Integration.countDocuments({ installationId: String(first.installation._id) })).toBe(1); + }); + + it('returns 202 for a fresh activating split, then takes it over and mints once after the lease', async () => { + const { userId, podId } = ids(); + const targetId = new mongoose.Types.ObjectId(userId); + const parent = await InstallableInstallation.create({ + installableId: 'telegram', + installableVersion: '1.0.0', + targetType: 'user', + targetId, + scope: 'user', + installedBy: targetId, + installSource: 'direct', + status: 'activating', + claimId: 'owner-a', + claimedAt: new Date(), + }); + await Integration.create({ + installationId: String(parent._id), + installationClaimId: 'owner-a', + podId, + type: 'telegram', + status: 'pending', + createdBy: targetId, + isActive: false, + config: { liveRelay: true, linkedUserId: userId }, + }); + + const waiting = await install({ installableId: 'telegram', installedBy: userId, podId }); + expect(waiting.httpStatus).toBe(202); + expect(waiting.state).toBe('activating'); + + parent.claimedAt = new Date(Date.now() - 61_000); + await parent.save(); + const retryPodId = new mongoose.Types.ObjectId().toString(); + const recovered = await install({ installableId: 'telegram', installedBy: userId, podId: retryPodId }); + expect(recovered.httpStatus).toBe(200); + expect(recovered.integration.config.connectCode).toMatch(/^[a-f0-9]{32}$/); + expect(String(recovered.integration.podId)).toBe(retryPodId); + expect(await Integration.countDocuments({ installationId: String(parent._id) })).toBe(1); + }); + + it('returns a typed lock loss when an activating parent has no projection', async () => { + const { userId, podId } = ids(); + const targetId = new mongoose.Types.ObjectId(userId); + await InstallableInstallation.create({ + installableId: 'telegram', + installableVersion: '1.0.0', + targetType: 'user', + targetId, + scope: 'user', + installedBy: targetId, + installSource: 'direct', + status: 'activating', + claimId: 'missing-projection-owner', + claimedAt: new Date(Date.now() - 61_000), + }); + + await expect(install({ installableId: 'telegram', installedBy: userId, podId })) + .rejects.toBeInstanceOf(InstallLockLostError); + expect(await Integration.countDocuments({ type: 'telegram' })).toBe(0); + }); + + it('refuses a revived stale owner without changing the winner code or unprojecting', async () => { + const { userId, podId } = ids(); + const originalFindOneAndUpdate = Integration.findOneAndUpdate.bind(Integration); + let releaseOwner; + let reachedOwnerActivation; + const ownerActivationReached = new Promise((resolve) => { reachedOwnerActivation = resolve; }); + const ownerMayContinue = new Promise((resolve) => { releaseOwner = resolve; }); + let firstActivation = true; + const findOneAndUpdate = jest.spyOn(Integration, 'findOneAndUpdate').mockImplementation(async (filter, ...args) => { + if (firstActivation && filter?.isActive === false && filter?.revokedAt?.$exists === false) { + firstActivation = false; + reachedOwnerActivation(); + await ownerMayContinue; + } + return originalFindOneAndUpdate(filter, ...args); + }); + const webhookProjector = projectorRegistry.get('webhook'); + const unproject = jest.spyOn(webhookProjector, 'unproject'); + + try { + const ownerA = install({ installableId: 'telegram', installedBy: userId, podId }); + await ownerActivationReached; + const parent = await InstallableInstallation.findOne({ installableId: 'telegram' }); + parent.claimedAt = new Date(Date.now() - 61_000); + await parent.save(); + + const ownerB = await install({ installableId: 'telegram', installedBy: userId, podId }); + const winnerCode = ownerB.integration.config.connectCode; + releaseOwner(null); + + await expect(ownerA).rejects.toBeInstanceOf(InstallLockLostError); + const finalParent = await InstallableInstallation.findById(parent._id); + const finalIntegration = await Integration.findOne({ installationId: String(parent._id) }); + expect(finalParent.status).toBe('active'); + expect(finalIntegration.config.connectCode).toBe(winnerCode); + expect(await Integration.countDocuments({ installationId: String(parent._id) })).toBe(1); + expect(unproject).not.toHaveBeenCalled(); + } finally { + findOneAndUpdate.mockRestore(); + unproject.mockRestore(); + } + }); + + it('uninstalls by identity, clears the code, and creates a new projection on re-install', async () => { + const { userId, podId } = ids(); + const first = await install({ installableId: 'telegram', installedBy: userId, podId }); + const integration = await Integration.findById(first.integration._id); + integration.config.relayMap = [{ tgMessageId: '7', agentUsername: 'kai' }]; + await integration.save(); + + const removed = await uninstall({ installableId: 'telegram', installedBy: userId }); + const inactive = await Integration.findById(first.integration._id); + expect(removed.status).toBe('uninstalled'); + expect(inactive.isActive).toBe(false); + expect(inactive.config.connectCode).toBeUndefined(); + expect(inactive.config.relayMap).toHaveLength(1); + expect(inactive.revokedAt).toBeInstanceOf(Date); + + const replacement = await install({ installableId: 'telegram', installedBy: userId, podId }); + expect(String(replacement.installation._id)).not.toBe(String(first.installation._id)); + expect(String(replacement.integration._id)).not.toBe(String(first.integration._id)); + const duplicateAfterHistoricalUninstall = await install({ + installableId: 'telegram', installedBy: userId, podId, + }); + expect(duplicateAfterHistoricalUninstall.httpStatus).toBe(200); + expect(String(duplicateAfterHistoricalUninstall.installation._id)).toBe( + String(replacement.installation._id), + ); + + await uninstall({ installableId: 'telegram', installedBy: userId }); + const replacementAfterSecondUninstall = await Integration.findById(replacement.integration._id); + expect(replacementAfterSecondUninstall.isActive).toBe(false); + }); + + it('finishes a stale revocation before re-installing a new parent and projection', async () => { + const { userId, podId } = ids(); + const first = await install({ installableId: 'telegram', installedBy: userId, podId }); + const oldIntegration = await Integration.findById(first.integration._id); + const oldCode = oldIntegration.config.connectCode; + const stale = await InstallableInstallation.findById(first.installation._id); + stale.status = 'uninstalling'; + stale.claimId = 'dead-uninstall-owner'; + stale.claimedAt = new Date(Date.now() - 61_000); + await stale.save(); + + const replacement = await install({ installableId: 'telegram', installedBy: userId, podId }); + const retired = await InstallableInstallation.findById(first.installation._id); + const deactivated = await Integration.findById(first.integration._id); + + expect(replacement.httpStatus).toBe(200); + expect(String(replacement.installation._id)).not.toBe(String(first.installation._id)); + expect(String(replacement.integration._id)).not.toBe(String(first.integration._id)); + expect(retired.status).toBe('uninstalled'); + expect(deactivated.isActive).toBe(false); + expect(deactivated.config.connectCode).toBeUndefined(); + expect(deactivated.revokedAt).toBeInstanceOf(Date); + expect(replacement.integration.isActive).toBe(true); + expect(replacement.integration.config.connectCode).toMatch(/^[a-f0-9]{32}$/); + expect(replacement.integration.config.connectCode).not.toBe(oldCode); + }); + + it('keeps a failed revocation recoverable until its projection is inactive', async () => { + const { userId, podId } = ids(); + const installed = await install({ installableId: 'telegram', installedBy: userId, podId }); + const originalEventProjector = projectorRegistry.get('event-handler'); + projectorRegistry.set('event-handler', { + ...originalEventProjector, + unproject: jest.fn().mockRejectedValue(new Error('projection unavailable')), + }); + + try { + await expect(uninstall({ installableId: 'telegram', installedBy: userId })) + .rejects.toThrow('projection unavailable'); + } finally { + projectorRegistry.set('event-handler', originalEventProjector); + } + + const pending = await InstallableInstallation.findById(installed.installation._id); + const inactive = await Integration.findById(installed.integration._id); + expect(pending.status).toBe('uninstalling'); + expect(inactive.isActive).toBe(false); + expect(inactive.config.connectCode).toBeUndefined(); + expect(inactive.revokedAt).toBeInstanceOf(Date); + const staleActivation = await Integration.findOneAndUpdate( + { + _id: inactive._id, + isActive: false, + revokedAt: { $exists: false }, + }, + { $set: { isActive: true } }, + { new: true }, + ); + expect(staleActivation).toBeNull(); + + pending.claimedAt = new Date(Date.now() - 61_000); + await pending.save(); + const reconciled = await sweep(new Date()); + expect(reconciled.uninstallsCompleted).toBe(1); + expect((await InstallableInstallation.findById(pending._id)).status).toBe('uninstalled'); + }); + + it('sweeps a crashed revocation by deactivating before it finalizes the parent', async () => { + const { userId, podId } = ids(); + const installed = await install({ installableId: 'telegram', installedBy: userId, podId }); + const parent = await InstallableInstallation.findById(installed.installation._id); + parent.status = 'uninstalling'; + parent.claimId = 'revocation-generation'; + parent.claimedAt = new Date(Date.now() - 61_000); + await parent.save(); + + const reconciled = await sweep(new Date()); + const finished = await InstallableInstallation.findById(parent._id); + const integration = await Integration.findById(installed.integration._id); + expect(reconciled.uninstallsCompleted).toBe(1); + expect(finished.status).toBe('uninstalled'); + expect(integration.isActive).toBe(false); + expect(integration.config.connectCode).toBeUndefined(); + expect(integration.installationClaimId).toBe(installed.installation.claimId); + expect(integration.revokedAt).toBeInstanceOf(Date); + }); + + it('sweeps stale installing and inactive activating rows to error', async () => { + const { userId, podId } = ids(); + const targetId = new mongoose.Types.ObjectId(userId); + const installing = await InstallableInstallation.create({ + installableId: 'telegram', installableVersion: '1.0.0', targetType: 'user', targetId, + scope: 'user', installedBy: targetId, installSource: 'direct', status: 'installing', + claimId: 'installing-a', claimedAt: new Date(Date.now() - 61_000), + }); + const activating = await InstallableInstallation.create({ + installableId: 'telegram-inactive', installableVersion: '1.0.0', targetType: 'user', + targetId: new mongoose.Types.ObjectId(), scope: 'user', installedBy: targetId, + installSource: 'direct', status: 'activating', claimId: 'activating-a', + claimedAt: new Date(Date.now() - 61_000), + }); + await Integration.create({ + installationId: String(activating._id), installationClaimId: 'activating-a', podId, + type: 'telegram', status: 'pending', createdBy: targetId, isActive: false, + config: { liveRelay: true, linkedUserId: userId }, + }); + + const reconciled = await sweep(new Date()); + expect(reconciled.errored).toBe(2); + expect((await InstallableInstallation.findById(installing._id)).status).toBe('error'); + expect((await InstallableInstallation.findById(activating._id)).status).toBe('error'); + }); + + it('marks a live parent stale when its projected Integration has been deleted', async () => { + const { userId, podId } = ids(); + const installed = await install({ installableId: 'telegram', installedBy: userId, podId }); + await Integration.deleteOne({ _id: installed.integration._id }); + + const reconciled = await sweep(new Date()); + const parent = await InstallableInstallation.findById(installed.installation._id); + expect(reconciled.staleComponents).toBe(1); + expect(parent.components.every((component) => component.status === 'stale')).toBe(true); + }); + + it('completes a stale activating row with its already-active, same-generation projection', async () => { + const { userId, podId } = ids(); + const result = await install({ installableId: 'telegram', installedBy: userId, podId }); + const installation = await InstallableInstallation.findById(result.installation._id); + const integration = await Integration.findById(result.integration._id); + const originalCode = integration.config.connectCode; + + installation.status = 'activating'; + installation.claimId = 'takeover-b'; + installation.claimedAt = new Date(Date.now() - 61_000); + await installation.save(); + integration.installationClaimId = 'owner-a'; + await integration.save(); + + const reconciled = await sweep(new Date()); + const completed = await InstallableInstallation.findById(installation._id); + const unchanged = await Integration.findById(integration._id); + expect(reconciled.completed).toBe(1); + expect(completed.status).toBe('active'); + expect(unchanged.config.connectCode).toBe(originalCode); + }); +}); diff --git a/backend/middleware/integrationRateLimit.ts b/backend/middleware/integrationRateLimit.ts new file mode 100644 index 000000000..4438d69f7 --- /dev/null +++ b/backend/middleware/integrationRateLimit.ts @@ -0,0 +1,46 @@ +import rateLimit, { ipKeyGenerator } from 'express-rate-limit'; +import { createHash } from 'crypto'; + +type RateLimitRequest = { + get?: (header: string) => string | undefined; + ip?: string; +}; + +// Connector writes can mint bearer connect codes. Keep the key and limits in +// one module so every route that creates or re-mints a connector shares the +// same bucket instead of each route being independently burstable. +export const integrationsRateLimitKey = (req: RateLimitRequest): string => { + const authHeader = req.get?.('authorization'); + if (authHeader) { + return `tok:${createHash('sha256').update(authHeader).digest('hex').slice(0, 16)}`; + } + return req.ip ? ipKeyGenerator(req.ip) : 'anon'; +}; + +export const writeIntegrationsRateLimit = rateLimit({ + windowMs: 60_000, + max: 30, + standardHeaders: true, + legacyHeaders: false, + keyGenerator: integrationsRateLimitKey, + handler: (_req: unknown, res: { status: (n: number) => { json: (body: unknown) => void } }) => { + res.status(429).json({ msg: 'rate limit exceeded: 30 writes per 60s' }); + }, +}); + +export const listIntegrationsRateLimit = rateLimit({ + windowMs: 60_000, + max: 120, + standardHeaders: true, + legacyHeaders: false, + keyGenerator: integrationsRateLimitKey, + handler: (_req: unknown, res: { status: (n: number) => { json: (body: unknown) => void } }) => { + res.status(429).json({ msg: 'rate limit exceeded: 120 reads per 60s' }); + }, +}); + +module.exports = { + integrationsRateLimitKey, + writeIntegrationsRateLimit, + listIntegrationsRateLimit, +}; diff --git a/backend/models/InstallableInstallation.ts b/backend/models/InstallableInstallation.ts index 347d979be..b8e3efcc8 100644 --- a/backend/models/InstallableInstallation.ts +++ b/backend/models/InstallableInstallation.ts @@ -30,13 +30,16 @@ import type { ComponentType, InstallableScope } from './Installable'; export type InstallationTargetType = 'pod' | 'user' | 'dm' | 'instance'; export type InstallationStatus = + | 'installing' + | 'activating' + | 'uninstalling' | 'active' | 'paused' | 'uninstalled' | 'error' | 'stale'; -export type InstallSource = 'marketplace' | 'registry' | 'direct' | 'system'; +export type InstallSource = 'marketplace' | 'registry' | 'direct' | 'ui' | 'system'; export interface IComponentInstallationUsage { lastUsedAt?: Date; @@ -92,6 +95,11 @@ export interface IInstallableInstallation extends Document { targetId: Types.ObjectId; scope: InstallableScope; // replicates Installable.scope for query perf + // The pod requested by the current claim. It makes a fresh transient legible + // before a projection exists; an errored or stale inactive projection may + // safely adopt the retry's pod at its fenced activation write. + boundPodId?: Types.ObjectId; + // Install provenance installedBy: Types.ObjectId; installSource: InstallSource; @@ -104,6 +112,13 @@ export interface IInstallableInstallation extends Document { // Lifecycle status: InstallationStatus; + /** + * Opaque ownership generation for a transient install claim. Every parent + * owner write includes it in its filter; a stale worker is then a no-op + * instead of overwriting the winner's lifecycle state or bearer code. + */ + claimId?: string; + claimedAt?: Date; errorMessage?: string; staleSince?: Date; @@ -176,11 +191,12 @@ const InstallableInstallationSchema = new Schema( enum: ['instance', 'pod', 'user', 'dm'], required: true, }, + boundPodId: { type: Schema.Types.ObjectId, ref: 'Pod' }, installedBy: { type: Schema.Types.ObjectId, ref: 'User', required: true }, installSource: { type: String, - enum: ['marketplace', 'registry', 'direct', 'system'], + enum: ['marketplace', 'registry', 'direct', 'ui', 'system'], required: true, }, @@ -190,9 +206,11 @@ const InstallableInstallationSchema = new Schema( status: { type: String, - enum: ['active', 'paused', 'uninstalled', 'error', 'stale'], + enum: ['installing', 'activating', 'uninstalling', 'active', 'paused', 'uninstalled', 'error', 'stale'], default: 'active', }, + claimId: { type: String }, + claimedAt: { type: Date }, errorMessage: { type: String }, staleSince: { type: Date }, }, @@ -203,10 +221,20 @@ const InstallableInstallationSchema = new Schema( // Indexes // --------------------------------------------------------------------------- -// One installation of a given Installable per target. +// One live installation of a given Installable per target. `uninstalled` +// rows intentionally fall outside the index: their audit history survives, +// while a later install gets a new parent/projection instead of resurrecting +// a historical connector. Retained `error` rows stay inside so retry can +// claim the same parent and safely reuse its inactive projection. InstallableInstallationSchema.index( { installableId: 1, targetType: 1, targetId: 1 }, - { unique: true }, + { + unique: true, + name: 'installable_live_target_unique', + partialFilterExpression: { + status: { $in: ['installing', 'activating', 'uninstalling', 'active', 'error'] }, + }, + }, ); // "What's installed in this pod/user/dm right now?" diff --git a/backend/models/Integration.ts b/backend/models/Integration.ts index a958fa330..b17793a1a 100644 --- a/backend/models/Integration.ts +++ b/backend/models/Integration.ts @@ -32,6 +32,10 @@ export interface IIntegrationMessageBuffer { export interface IIntegration extends Document { installationId?: string; + /** InstallableInstallation claim generation that minted this projection's code. */ + installationClaimId?: string; + /** Terminal tombstone: a revoked projection can never be activated again. */ + revokedAt?: Date; podId: Types.ObjectId; type: IntegrationType; status: IntegrationStatus; @@ -107,6 +111,8 @@ export interface IIntegration extends Document { const IntegrationSchema = new Schema( { installationId: { type: String, unique: true, sparse: true }, + installationClaimId: { type: String }, + revokedAt: { type: Date }, podId: { type: Schema.Types.ObjectId, ref: 'Pod', required: true }, type: { type: String, diff --git a/backend/routes/discord.ts b/backend/routes/discord.ts index 97a7286e1..798bf0f88 100644 --- a/backend/routes/discord.ts +++ b/backend/routes/discord.ts @@ -20,6 +20,9 @@ const User = require('../models/User'); const DiscordService = require('../services/discordService'); // eslint-disable-next-line global-require const { runDiscordCommandForIntegrations } = require('../services/discordMultiCommandService'); +// Static import keeps the destructive route visible to CodeQL's rate-limit +// query while sharing the connector write bucket with its sibling routes. +import { writeIntegrationsRateLimit } from '../middleware/integrationRateLimit'; interface AuthReq { user?: { id: string; role?: string }; @@ -202,10 +205,14 @@ router.get('/binding/:podId', auth, async (req: AuthReq, res: Res) => { } }); -router.delete('/uninstall/:installationId', auth, async (req: AuthReq, res: Res) => { +router.delete('/uninstall/:installationId', writeIntegrationsRateLimit, auth, async (req: AuthReq, res: Res) => { try { const { installationId } = req.params || {}; - const integration = await Integration.findOne({ installationId }) as { _id: unknown; createdBy?: { toString: () => string }; podId?: unknown } | null; + // `installationId` is no longer Discord-owned: Installable app projections + // use it as their parent back-pointer too. Keep this destructive legacy + // route provider-scoped so a Discord installation id can never delete a + // connector row from another provider. + const integration = await Integration.findOne({ installationId, type: 'discord' }) as { _id: unknown; createdBy?: { toString: () => string }; podId?: unknown } | null; if (!integration) return res.status(404).json({ error: 'Integration not found' }); const canManage = await canManageIntegration(integration, req.user?.id || ''); if (!canManage) return res.status(403).json({ error: 'Access denied' }); diff --git a/backend/routes/installables.ts b/backend/routes/installables.ts new file mode 100644 index 000000000..c2d03c489 --- /dev/null +++ b/backend/routes/installables.ts @@ -0,0 +1,146 @@ +// eslint-disable-next-line global-require +const express = require('express'); +// eslint-disable-next-line global-require +const auth = require('../middleware/auth'); +// eslint-disable-next-line global-require +const Pod = require('../models/Pod'); +// eslint-disable-next-line global-require +const isPodMember = require('../utils/isPodMember'); +import { Types } from 'mongoose'; +import { writeIntegrationsRateLimit } from '../middleware/integrationRateLimit'; +// eslint-disable-next-line global-require +const { + InstallLockLostError, + InstallableAlreadyInstalledError, + InstallableNotFoundError, + InstallableProjectionError, + InstallInProgressError, + install, + uninstall, +} = require('../services/installable/installableInstallationService'); + +interface AuthReq { + user?: { id?: string; _id?: string }; + userId?: string; + params?: Record; + body?: Record; +} + +interface Res { + status: (status: number) => Res; + json: (body: unknown) => void; +} + +const router: ReturnType = express.Router(); + +const requesterId = (req: AuthReq): string | undefined => req.userId || req.user?.id || req.user?._id; + +const sendInstallError = (error: Error, res: Res): void => { + if (error instanceof InstallableNotFoundError) { + res.status(404).json({ code: 'installable_not_found', error: error.message }); + return; + } + if (error instanceof InstallLockLostError) { + res.status(409).json({ code: 'install_lock_lost', error: error.message }); + return; + } + if (error instanceof InstallInProgressError) { + const { boundPodId } = error as Error & { boundPodId?: string }; + res.status(409).json({ + code: 'install_in_progress', + error: error.message, + ...(boundPodId ? { boundPodId } : {}), + }); + return; + } + if (error instanceof InstallableAlreadyInstalledError) { + const { boundPodId } = error as Error & { boundPodId: string }; + res.status(409).json({ + code: 'already_installed', + error: error.message, + boundPodId, + }); + return; + } + if (error instanceof InstallableProjectionError) { + res.status(422).json({ code: 'install_projection_failed', error: error.message }); + return; + } + if (/must be a valid ObjectId/.test(error.message)) { + res.status(400).json({ error: error.message }); + return; + } + console.error('[installable] install failed:', error.message); + res.status(500).json({ error: 'Could not install connector' }); +}; + +/** + * POST /api/installables/:installableId/install + * + * Phase 1 is intentionally narrow: the only client choice is the pod that + * becomes this user-scoped connector's first gate. Target identity is always + * taken from auth, never from a body field. + */ +router.post('/:installableId/install', writeIntegrationsRateLimit, auth, async (req: AuthReq, res: Res) => { + const body = req.body || {}; + if (Array.isArray(body) || typeof body !== 'object') { + return res.status(400).json({ error: 'Body must be an object with podId' }); + } + if (Object.keys(body).some((key) => key !== 'podId')) { + return res.status(400).json({ error: 'Only podId is accepted when installing a connector' }); + } + const podId = body.podId; + if (typeof podId !== 'string' || !podId) { + return res.status(400).json({ error: 'podId is required' }); + } + if (!Types.ObjectId.isValid(podId)) { + return res.status(400).json({ error: 'podId must be a valid ObjectId' }); + } + const userId = requesterId(req); + if (!userId) return res.status(401).json({ error: 'Unauthorized' }); + + try { + const pod = await Pod.findById(podId); + if (!pod || !isPodMember(pod, userId)) { + return res.status(403).json({ error: 'Access denied' }); + } + const result = await install({ + installableId: String(req.params?.installableId || ''), + installedBy: String(userId), + podId, + }); + return res.status(result.httpStatus).json({ + status: result.state, + installation: result.installation, + integration: result.integration, + ...(result.boundPodId ? { boundPodId: result.boundPodId } : {}), + }); + } catch (error) { + return sendInstallError(error as Error, res); + } +}); + +/** + * DELETE /api/installables/:installableId/install + * + * There is no installation id in the path or body. The user's target row is + * the only row this endpoint can ever deactivate, even if they share its pod. + */ +router.delete('/:installableId/install', writeIntegrationsRateLimit, auth, async (req: AuthReq, res: Res) => { + const userId = requesterId(req); + if (!userId) return res.status(401).json({ error: 'Unauthorized' }); + try { + const installation = await uninstall({ + installableId: String(req.params?.installableId || ''), + installedBy: String(userId), + }); + if (installation.status === 'uninstalling') { + return res.status(202).json({ status: 'uninstalling', installation }); + } + return res.json({ status: 'uninstalled', installation }); + } catch (error) { + return sendInstallError(error as Error, res); + } +}); + +module.exports = router; diff --git a/backend/routes/integrations.ts b/backend/routes/integrations.ts index eb45b24c4..9ebf45b15 100644 --- a/backend/routes/integrations.ts +++ b/backend/routes/integrations.ts @@ -1,9 +1,5 @@ // eslint-disable-next-line global-require const express = require('express'); -// ESM import (not require) so CodeQL's js/missing-rate-limiting query -// recognizes the limiter (same pattern as routes/messages.ts). -import rateLimit, { ipKeyGenerator } from 'express-rate-limit'; -import { createHash } from 'crypto'; // eslint-disable-next-line global-require const axios = require('axios'); // eslint-disable-next-line global-require @@ -34,6 +30,12 @@ const { hash, randomSecret } = require('../utils/secret'); const { mintConnectCode } = require('../services/telegramConnectCode'); // eslint-disable-next-line global-require const isPodMember = require('../utils/isPodMember'); +// Keep this as an ESM import: static analysis recognizes the rate limiter at +// the route sink, while the middleware owns the shared token/IP bucket. +import { + writeIntegrationsRateLimit, + listIntegrationsRateLimit, +} from '../middleware/integrationRateLimit'; // Bridge attribution + binding fields are server-owned. linkedUserId is the // identity every inbound live-relay message is AUTHORED as; chatId/chatType @@ -269,29 +271,6 @@ router.get('/:podId', auth, async (req: AuthReq, res: Res) => { } }); -// Token/IP keying shared by every limiter in this file — same shape as -// routes/messages.ts so NAT'd users don't share a bucket. -const integrationsRateLimitKey = (req: { get?: (h: string) => string | undefined; ip?: string }): string => { - const authHeader = req.get?.('authorization'); - if (authHeader) { - return `tok:${createHash('sha256').update(authHeader).digest('hex').slice(0, 16)}`; - } - return req.ip ? ipKeyGenerator(req.ip) : 'anon'; -}; - -// Write limiter for the create + re-mint paths: each one mints a connect code -// (a bearer secret) and writes a row, so a burst is either a bug or a probe. -const writeIntegrationsRateLimit = rateLimit({ - windowMs: 60_000, - max: 30, - standardHeaders: true, - legacyHeaders: false, - keyGenerator: integrationsRateLimitKey, - handler: (_req: unknown, res: { status: (n: number) => { json: (b: unknown) => void } }) => { - res.status(429).json({ msg: 'rate limit exceeded: 30 writes per 60s' }); - }, -}); - router.post('/', writeIntegrationsRateLimit, auth, async (req: AuthReq, res: Res) => { try { const { podId, type, config } = (req.body || {}) as { podId?: string; type?: string; config?: Record }; @@ -466,19 +445,6 @@ router.get('/admin/all', auth, adminAuth, async (_req: AuthReq, res: Res) => { } }); -// Read limiter for the connector listing — same token/IP keying as -// routes/messages.ts so NAT'd users don't share a bucket. -const listIntegrationsRateLimit = rateLimit({ - windowMs: 60_000, - max: 120, - standardHeaders: true, - legacyHeaders: false, - keyGenerator: integrationsRateLimitKey, - handler: (_req: unknown, res: { status: (n: number) => { json: (b: unknown) => void } }) => { - res.status(429).json({ msg: 'rate limit exceeded: 120 reads per 60s' }); - }, -}); - router.get('/user/all', listIntegrationsRateLimit, auth, async (req: AuthReq, res: Res) => { try { const integrations = await Integration.find({ createdBy: req.user?.id, isActive: true }).populate('podId', 'name type').sort({ createdAt: -1 }); diff --git a/backend/scripts/seed-builtin-connectors.ts b/backend/scripts/seed-builtin-connectors.ts new file mode 100644 index 000000000..03ac1a435 --- /dev/null +++ b/backend/scripts/seed-builtin-connectors.ts @@ -0,0 +1,86 @@ +// Builtin connector manifests are catalog entries, not auto-installs. The +// seeder makes Telegram available at boot but never creates an installation or +// bearer connect code until a user explicitly chooses "Add a channel". + +// eslint-disable-next-line @typescript-eslint/no-require-imports, global-require +const Installable = require('../models/Installable'); +// eslint-disable-next-line @typescript-eslint/no-require-imports, global-require +const InstallableInstallation = require('../models/InstallableInstallation'); +// eslint-disable-next-line @typescript-eslint/no-require-imports, global-require +const { manifests } = require('../integrations/manifests'); + +const telegramCatalog = manifests.telegram?.catalog; +if (!telegramCatalog) { + throw new Error('Telegram provider manifest is missing its catalog metadata'); +} + +const TELEGRAM_CONNECTOR = { + installableId: 'telegram', + // The provider registry owns catalog display copy. The Installable is the + // package wrapper, so it must not fork a second source of truth. + name: telegramCatalog.label, + description: telegramCatalog.description, + version: '1.0.0', + kind: 'app', + source: 'builtin', + scope: 'user', + status: 'active', + requires: ['chat:read', 'chat:write', 'integrations:manage'], + components: [ + { + name: 'telegram-webhook', + type: 'webhook', + webhookPath: '/api/webhooks/telegram', + webhookEvents: ['message', 'edited_message'], + addresses: [{ mode: 'webhook', identifier: '/api/webhooks/telegram' }], + scopes: ['chat:write'], + }, + { + name: 'telegram-relay', + type: 'event-handler', + eventType: 'chat.message', + eventHandler: 'internal:telegram.relay', + addresses: [{ mode: 'event', identifier: 'chat.message' }], + scopes: ['chat:read'], + }, + ], +}; + +const migrateInstallationIndex = async (): Promise => { + // The scaffolding index was unique across historical uninstalls. D17 makes + // a new installation a new projection, so replace it exactly once with the + // partial live-state index declared on the model. + const indexes = await InstallableInstallation.collection.indexes(); + const legacy = indexes.find((index: { name?: string; partialFilterExpression?: unknown }) => ( + index.name === 'installableId_1_targetType_1_targetId_1' + && !index.partialFilterExpression + )); + if (legacy?.name) { + await InstallableInstallation.collection.dropIndex(legacy.name); + console.log('[builtin-connectors] replaced legacy installation uniqueness index'); + } + await InstallableInstallation.syncIndexes(); +}; + +export const seedBuiltinConnectors = async (): Promise => { + try { + await migrateInstallationIndex(); + await Installable.findOneAndUpdate( + { installableId: TELEGRAM_CONNECTOR.installableId }, + { + $set: TELEGRAM_CONNECTOR, + $setOnInsert: { + stats: { totalInstalls: 0, activeInstalls: 0, forkCount: 0 }, + }, + }, + { upsert: true, new: true, setDefaultsOnInsert: true }, + ); + console.log('[builtin-connectors] Telegram manifest ready'); + } catch (error) { + console.error('[builtin-connectors] seed failed:', (error as Error).message); + } +}; + +export { TELEGRAM_CONNECTOR }; + +module.exports = { seedBuiltinConnectors, TELEGRAM_CONNECTOR }; diff --git a/backend/server.ts b/backend/server.ts index b10c8f59e..4a7e73793 100644 --- a/backend/server.ts +++ b/backend/server.ts @@ -200,6 +200,7 @@ app.use('/api/uploads', uploadsRoutes); app.use('/api/docs', docsRoutes); app.use('/api/summaries', summariesRoutes); app.use('/api/integrations', integrationRoutes); +app.use('/api/installables', require('./routes/installables')); app.use('/api/apps', appPlatformRoutes); app.use('/api/webhooks/discord', discordWebhookRoutes); app.use('/api/webhooks/slack', slackWebhookRoutes); @@ -292,6 +293,11 @@ mongoose.connection.once('open', () => { require('./scripts/seed-native-agents').seedNativeAgents().catch((err: any) => console.error('[native-seed] failed:', err?.message || err), ); + require('./scripts/seed-builtin-connectors').seedBuiltinConnectors().then(() => + require('./services/installable/installableReconciler').sweep(), + ).catch((err: any) => + console.error('[builtin-connectors] bootstrap failed:', err?.message || err), + ); })(); } }); diff --git a/backend/services/agentMessageService.ts b/backend/services/agentMessageService.ts index 82ec4bd66..a37777e80 100644 --- a/backend/services/agentMessageService.ts +++ b/backend/services/agentMessageService.ts @@ -1778,15 +1778,15 @@ class AgentMessageService { console.error('Failed to emit agent socket message:', socketError); } - // Telegram live bridge (fire-and-forget): pods with a live-relay telegram - // integration surface escalations and lead reports into the linked chat. - // The bridge no-ops in O(1 query) for every pod without one, and a bridge - // failure never fails the post. + // Installable event dispatch (fire-and-forget). Selection is pod-scoped in + // the dispatcher before a connector handler runs, so one user's relay can + // never observe another user's event. A handler failure never fails the + // message post. try { // eslint-disable-next-line global-require, @typescript-eslint/no-require-imports - const bridge = require('./telegramBridgeService'); + const eventHandlers = require('./installable/eventHandlers'); const bridgeUsername = AgentIdentityService.buildAgentUsername(agentName, instanceId); - void bridge.relayAgentMessageToTelegram({ + void eventHandlers.dispatch('chat.message', { podId: String(podId), agentUsername: bridgeUsername, displayName: senderDisplayName || bridgeUsername, diff --git a/backend/services/installable/eventHandlers.ts b/backend/services/installable/eventHandlers.ts new file mode 100644 index 000000000..d36829230 --- /dev/null +++ b/backend/services/installable/eventHandlers.ts @@ -0,0 +1,152 @@ +import type { IIntegration } from '../../models/Integration'; +import { Types } from 'mongoose'; + +// eslint-disable-next-line @typescript-eslint/no-require-imports, global-require +const Integration = require('../../models/Integration'); +// eslint-disable-next-line @typescript-eslint/no-require-imports, global-require +const telegramBridgeService = require('../telegramBridgeService'); + +export interface ChatMessageEventPayload { + podId: string; + agentUsername: string; + displayName: string; + content: string; + podMessageId?: string | null; +} + +export type InternalEventHandler = ( + payload: ChatMessageEventPayload & { integration: IIntegration }, +) => Promise; + +// The registry is intentionally a mutable object rather than a switch so +// component tests can pin selection at this seam and future component types +// add one registration instead of another dispatcher branch. +export const eventHandlers: Record = { + 'telegram.relay': telegramBridgeService.relayAgentMessageToTelegram, +}; + +export const hasEventHandler = (reference: string): boolean => ( + reference.startsWith('internal:') + && typeof eventHandlers[reference.slice('internal:'.length)] === 'function' +); + +const activeTelegramHandlersForPod = async (podId: string): Promise> => { + // Selection lives here, before any handler invocation. The $lookup keeps the + // event path O(1) and proves the target pod's connector is the only one + // eligible; individual bridges remain defensive, not authoritative. + if (!Types.ObjectId.isValid(podId)) return []; + return Integration.aggregate([ + { + $match: { + type: 'telegram', + isActive: true, + podId: new Types.ObjectId(podId), + 'config.liveRelay': true, + }, + }, + { + $lookup: { + from: 'installableinstallations', + let: { installationId: '$installationId' }, + pipeline: [ + { + $match: { + $expr: { $eq: [{ $toString: '$_id' }, '$$installationId'] }, + }, + }, + { + $match: { + status: 'active', + components: { + $elemMatch: { + componentType: 'event-handler', + status: 'active', + 'config.eventType': 'chat.message', + }, + }, + }, + }, + ], + as: 'installation', + }, + }, + { + $project: { + integration: '$$ROOT', + // Existing direct Integration rows predate D17. They remain one + // pod-scoped Telegram relay until their owner explicitly replaces or + // uninstalls them; only rows carrying a parent must prove an active + // component projection through the lookup. + handlers: { + $cond: [ + { $eq: [{ $ifNull: ['$installationId', null] }, null] }, + [{ config: { eventHandler: 'internal:telegram.relay' } }], + { + $let: { + vars: { parent: { $arrayElemAt: ['$installation', 0] } }, + in: { + $cond: [ + { $eq: ['$$parent.status', 'active'] }, + { + $filter: { + input: '$$parent.components', + as: 'component', + cond: { + $and: [ + { $eq: ['$$component.componentType', 'event-handler'] }, + { $eq: ['$$component.status', 'active'] }, + { $eq: ['$$component.config.eventType', 'chat.message'] }, + ], + }, + }, + }, + [], + ], + }, + }, + }, + ], + }, + }, + }, + { $unwind: '$handlers' }, + { $project: { integration: 1, handler: '$handlers.config.eventHandler' } }, + ]); +}; + +export const dispatch = async ( + eventType: 'chat.message', + payload: ChatMessageEventPayload, +): Promise => { + if (eventType !== 'chat.message') return; + + let selected: Array<{ integration: IIntegration; handler: string }>; + try { + selected = await activeTelegramHandlersForPod(payload.podId); + } catch (error) { + console.warn('[installable-dispatch] selector failed:', (error as Error).message); + return; + } + + await Promise.all(selected.map(async ({ integration, handler }) => { + const callback = eventHandlers[String(handler).replace(/^internal:/, '')]; + if (!callback) { + console.warn('[installable-dispatch] unregistered handler:', handler); + return; + } + try { + await callback({ ...payload, integration }); + } catch (error) { + console.warn('[installable-dispatch] handler failed:', (error as Error).message); + } + })); +}; + +module.exports = { + eventHandlers, + hasEventHandler, + dispatch, +}; diff --git a/backend/services/installable/installableInstallationService.ts b/backend/services/installable/installableInstallationService.ts new file mode 100644 index 000000000..6bd04d04f --- /dev/null +++ b/backend/services/installable/installableInstallationService.ts @@ -0,0 +1,658 @@ +import { randomUUID } from 'crypto'; +import { Types } from 'mongoose'; + +import type { IComponent, IInstallable } from '../../models/Installable'; +import type { + IComponentInstallation, + IInstallableInstallation, + InstallationStatus, +} from '../../models/InstallableInstallation'; +import { mintConnectCode } from '../telegramConnectCode'; +import { getProjector } from './projectors'; +import type { ProjectionIds } from './projectors/types'; + +// eslint-disable-next-line @typescript-eslint/no-require-imports, global-require +const Installable = require('../../models/Installable'); +// eslint-disable-next-line @typescript-eslint/no-require-imports, global-require +const InstallableInstallation = require('../../models/InstallableInstallation'); +// eslint-disable-next-line @typescript-eslint/no-require-imports, global-require +const Integration = require('../../models/Integration'); + +export const INSTALL_LOCK_TTL_MS = 60_000; + +export class InstallLockLostError extends Error { + code = 'install_lock_lost'; + + constructor() { + super('This install attempt was superseded by a newer attempt.'); + this.name = 'InstallLockLostError'; + } +} + +export class InstallableProjectionError extends Error { + code = 'install_projection_failed'; + + constructor(message: string) { + super(message); + this.name = 'InstallableProjectionError'; + } +} + +export class InstallInProgressError extends Error { + code = 'install_in_progress'; + boundPodId?: string; + + constructor(boundPodId?: string) { + super('This install is still in progress; try again shortly.'); + this.name = 'InstallInProgressError'; + this.boundPodId = boundPodId; + } +} + +export class InstallableAlreadyInstalledError extends Error { + code = 'already_installed'; + boundPodId: string; + + constructor(boundPodId: string) { + super('This connector is already installed for another pod.'); + this.name = 'InstallableAlreadyInstalledError'; + this.boundPodId = boundPodId; + } +} + +export class InstallableNotFoundError extends Error { + code = 'installable_not_found'; + + constructor() { + super('Installable not found'); + this.name = 'InstallableNotFoundError'; + } +} + +export interface InstallResult { + installation: IInstallableInstallation; + integration: unknown | null; + httpStatus: 200 | 202; + state: 'active' | 'installing' | 'activating' | 'uninstalling'; + boundPodId?: string; +} + +interface ClaimResult { + installation: IInstallableInstallation; + ownsClaim: boolean; + claimedState: 'installing' | 'activating' | 'uninstalling' | 'active'; +} + +interface InstallArgs { + installableId: string; + installedBy: string; + podId: string; +} + +const installationKeys = (installableId: string, targetId: Types.ObjectId) => ({ + installableId, + targetType: 'user', + targetId, +}); + +const liveClaimStatuses: InstallationStatus[] = [ + 'installing', + 'activating', + 'uninstalling', + 'active', + 'error', +]; + +const asObjectId = (value: string, field: string): Types.ObjectId => { + if (!Types.ObjectId.isValid(value)) throw new Error(`${field} must be a valid ObjectId`); + return new Types.ObjectId(value); +}; + +const statusForClaim = ( + status: InstallationStatus | undefined, +): ClaimResult['claimedState'] => { + if (status === 'active') return 'active'; + if (status === 'activating') return 'activating'; + if (status === 'uninstalling') return 'uninstalling'; + return 'installing'; +}; + +const isStale = (claimedAt: Date | undefined, now: Date): boolean => ( + !claimedAt || claimedAt.getTime() <= now.getTime() - INSTALL_LOCK_TTL_MS +); + +const hasClaim = (installation: IInstallableInstallation, claimId: string): boolean => ( + installation.claimId === claimId +); + +const projectionIdsToObject = (ids: ProjectionIds): Map => { + const entries = Object.entries(ids).map(([key, value]) => [ + key, + value instanceof Types.ObjectId ? value : new Types.ObjectId(String(value)), + ] as [string, Types.ObjectId]); + return new Map(entries); +}; + +const componentRecord = ( + component: IComponent, + projectionIds: ProjectionIds, +): IComponentInstallation => ({ + componentName: component.name, + componentType: component.type, + instanceId: 'default', + status: 'active', + config: new Map([ + ['eventType', component.eventType || ''], + ['eventHandler', component.eventHandler || ''], + ]), + projectionIds: projectionIdsToObject(projectionIds), + usage: { totalCalls: 0 }, + createdAt: new Date(), + updatedAt: new Date(), +}); + +const integrationFor = async (installation: IInstallableInstallation): Promise => ( + Integration.findOne({ installationId: String(installation._id) }) +); + +const integrationPodId = (integration: unknown): string | null => { + if (!integration || typeof integration !== 'object') return null; + const podId = (integration as { podId?: unknown }).podId; + return podId == null ? null : String(podId); +}; + +const installationBoundPodId = (installation: IInstallableInstallation): string | null => ( + installation.boundPodId == null ? null : String(installation.boundPodId) +); + +const resultForExisting = async ( + installation: IInstallableInstallation, + requestedPodId: Types.ObjectId, +): Promise => { + const integration = await integrationFor(installation); + if (installation.status === 'active') { + // A live Integration is a real channel binding. Every non-active state is + // retryable: its inactive projection has no code and activation will bind + // it to the retry's requested pod under the same CAS that mints the code. + // The Integration is the routing authority once it exists. `boundPodId` + // records claim intent and can be newer than a projection after a crash + // between activation and parent completion; reporting that intent first + // would tell a caller that the channel is bound to a pod it does not + // actually route to. + const boundPodId = integrationPodId(integration) || installationBoundPodId(installation); + if (boundPodId && boundPodId !== String(requestedPodId)) { + throw new InstallableAlreadyInstalledError(boundPodId); + } + return { installation, integration, httpStatus: 200, state: 'active' }; + } + // While a claim is fresh, its parent records the only durable target. An + // Integration may be an inactive projection left by an earlier attempt, so + // it must not decide this in-flight install's destination. A differently + // targeted caller needs the pending pod, not a success-shaped 202 that + // later surfaces a connector somewhere else. + const boundPodId = installationBoundPodId(installation); + if (boundPodId && boundPodId !== String(requestedPodId)) { + throw new InstallInProgressError(boundPodId); + } + return { + installation, + integration, + httpStatus: 202, + state: installation.status === 'activating' + ? 'activating' + : installation.status === 'uninstalling' + ? 'uninstalling' + : 'installing', + boundPodId: boundPodId || undefined, + }; +}; + +const claimInstallation = async ( + installable: IInstallable, + targetId: Types.ObjectId, + installedBy: Types.ObjectId, + requestedPodId: Types.ObjectId, + retried = false, +): Promise => { + const now = new Date(); + const staleBefore = new Date(now.getTime() - INSTALL_LOCK_TTL_MS); + const claimId = randomUUID(); + const keys = installationKeys(installable.installableId, targetId); + + // This is the installation lock. An existing active/fresh transient row is + // deliberately excluded from the filter; its unique partial index turns an + // attempted upsert into a duplicate-key loser path below, which then returns + // the winner without ever running a projector. + try { + const installation = await InstallableInstallation.findOneAndUpdate( + { + ...keys, + $or: [ + { status: 'error' }, + { status: { $in: ['installing', 'activating'] }, claimedAt: { $lte: staleBefore } }, + { status: 'uninstalling', claimedAt: { $lte: staleBefore } }, + { status: { $exists: false } }, + ], + }, + [ + { + $set: { + installableVersion: installable.version, + scope: installable.scope, + installedBy, + installSource: 'ui', + grantedScopes: installable.requires || [], + status: { + $cond: [ + { $eq: ['$status', 'activating'] }, + 'activating', + { + $cond: [ + { $eq: ['$status', 'uninstalling'] }, + 'uninstalling', + 'installing', + ], + }, + ], + }, + claimId, + claimedAt: now, + boundPodId: requestedPodId, + errorMessage: null, + components: { $ifNull: ['$components', []] }, + }, + }, + ], + { new: true, upsert: true }, + ) as IInstallableInstallation; + + return { + installation, + ownsClaim: hasClaim(installation, claimId), + claimedState: statusForClaim(installation.status), + }; + } catch (error) { + if ((error as { code?: number }).code !== 11000) throw error; + + const winner = await InstallableInstallation.findOne({ + ...keys, + status: { $in: liveClaimStatuses }, + }) as IInstallableInstallation | null; + if (!winner) throw error; + if (( + winner.status === 'error' + || ( + (winner.status === 'installing' + || winner.status === 'activating' + || winner.status === 'uninstalling') + && isStale(winner.claimedAt, now) + ) + ) && !retried) { + // A concurrent state transition beat our filter. Retrying the atomic + // claim is safe; only a returned matching generation owns work. + return claimInstallation(installable, targetId, installedBy, requestedPodId, true); + } + return { + installation: winner, + ownsClaim: false, + claimedState: statusForClaim(winner.status), + }; + } +}; + +const throwIfLockLost = (installation: IInstallableInstallation | null): IInstallableInstallation => { + if (!installation) throw new InstallLockLostError(); + return installation; +}; + +const markProjectionFailure = async ( + installation: IInstallableInstallation, + claimId: string, + components: IComponentInstallation[], + error: Error, +): Promise => { + const failed = await InstallableInstallation.findOneAndUpdate( + { _id: installation._id, status: 'installing', claimId }, + { + $set: { + status: 'error', + errorMessage: error.message, + components, + }, + }, + { new: true }, + ) as IInstallableInstallation | null; + throwIfLockLost(failed); +}; + +const projectComponents = async ( + installation: IInstallableInstallation, + installable: IInstallable, + installedBy: Types.ObjectId, + podId: Types.ObjectId, + claimId: string, +): Promise => { + const components: IComponentInstallation[] = []; + try { + for (const component of installable.components) { + const projector = getProjector(component.type); + if (!projector) throw new Error(`Unsupported component type: ${component.type}`); + const projectionIds = await projector.project(component, { + installation, + installable, + installedBy, + podId, + claimId, + }); + components.push(componentRecord(component, projectionIds)); + } + } catch (error) { + await markProjectionFailure(installation, claimId, components, error as Error); + throw new InstallableProjectionError((error as Error).message); + } + + const projected = await InstallableInstallation.findOneAndUpdate( + { _id: installation._id, status: 'installing', claimId }, + { $set: { components } }, + { new: true }, + ) as IInstallableInstallation | null; + return throwIfLockLost(projected); +}; + +const transitionToActivating = async ( + installation: IInstallableInstallation, + claimId: string, +): Promise => { + if (installation.status === 'activating') return installation; + const activating = await InstallableInstallation.findOneAndUpdate( + { _id: installation._id, status: 'installing', claimId }, + { $set: { status: 'activating', claimedAt: new Date() } }, + { new: true }, + ) as IInstallableInstallation | null; + return throwIfLockLost(activating); +}; + +const renewActivationLease = async ( + installation: IInstallableInstallation, + claimId: string, +): Promise => { + // This is the last parent CAS before the cross-collection activation write. + // If a stalled owner was taken over, it observes a null result here and does + // absolutely nothing to the winner's projection. If it wins this renewal, + // no valid takeover can claim the parent for the following lease window. + const renewed = await InstallableInstallation.findOneAndUpdate( + { _id: installation._id, status: 'activating', claimId }, + { $set: { claimedAt: new Date() } }, + { new: true }, + ) as IInstallableInstallation | null; + return throwIfLockLost(renewed); +}; + +const activateIntegration = async ( + installation: IInstallableInstallation, + targetPodId: Types.ObjectId, + claimId: string, +): Promise => { + const eligible = await Integration.exists({ + installationId: String(installation._id), + isActive: false, + revokedAt: { $exists: false }, + }); + if (!eligible) { + const existing = await integrationFor(installation) as { + isActive?: boolean; + podId?: unknown; + config?: { connectCode?: string }; + } | null; + if (existing?.isActive && existing.config?.connectCode) { + const boundPodId = integrationPodId(existing); + if (boundPodId && boundPodId !== String(targetPodId)) { + throw new InstallableAlreadyInstalledError(boundPodId); + } + return existing; + } + throw new InstallLockLostError(); + } + + const minted = mintConnectCode(); + const activated = await Integration.findOneAndUpdate( + { + installationId: String(installation._id), + isActive: false, + revokedAt: { $exists: false }, + }, + { + $set: { + isActive: true, + // An inactive projection has no redeemable code or chat binding. Move + // its target here, atomically with activation, so an error retry or + // stale takeover cannot report success for the prior request's pod. + podId: targetPodId, + installationClaimId: claimId, + 'config.connectCode': minted.connectCode, + 'config.connectCodeExpiresAt': minted.connectCodeExpiresAt, + }, + }, + { new: true }, + ); + if (activated) return activated; + + // A retry after a crash between writes 2 and 3 sees the already-minted + // code. It must complete the parent only; minting again invalidates the + // code the user may already have copied. + const existing = await integrationFor(installation) as { + isActive?: boolean; + podId?: unknown; + config?: { connectCode?: string }; + } | null; + if (existing?.isActive && existing.config?.connectCode) { + const boundPodId = integrationPodId(existing); + if (boundPodId && boundPodId !== String(targetPodId)) { + throw new InstallableAlreadyInstalledError(boundPodId); + } + return existing; + } + // A missing or revoked projection is not an internal 500: this claim no + // longer has an activation it may finish. Callers stop at the typed 409. + throw new InstallLockLostError(); +}; + +const finishActivation = async ( + installation: IInstallableInstallation, + claimId: string, +): Promise => { + const completed = await InstallableInstallation.findOneAndUpdate( + { _id: installation._id, status: 'activating', claimId }, + { $set: { status: 'active', errorMessage: null } }, + { new: true }, + ) as IInstallableInstallation | null; + return throwIfLockLost(completed); +}; + +const unprojectInstallation = async ( + installation: IInstallableInstallation, + installable: IInstallable | null, + installedBy: Types.ObjectId, + claimId: string, +): Promise => { + if (installable) { + for (const component of installable.components) { + const projector = getProjector(component.type); + if (!projector) continue; + const installedComponent = installation.components.find( + (entry) => entry.componentName === component.name, + ); + const ids = installedComponent?.projectionIds + ? Object.fromEntries(installedComponent.projectionIds.entries()) as ProjectionIds + : {}; + await projector.unproject(component, { + installation, + installable, + installedBy, + claimId, + }, ids); + } + return; + } + + // The parent is the authority. A retired manifest must not make a user + // unable to deactivate its projected channel row. + await Integration.findOneAndUpdate( + { installationId: String(installation._id) }, + { + $set: { + isActive: false, + revokedAt: new Date(), + }, + $unset: { + 'config.connectCode': 1, + 'config.connectCodeExpiresAt': 1, + }, + }, + ); +}; + +const finishUninstall = async ( + installation: IInstallableInstallation, + claimId: string, +): Promise => { + const uninstalled = await InstallableInstallation.findOneAndUpdate( + { _id: installation._id, status: 'uninstalling', claimId }, + { $set: { status: 'uninstalled', errorMessage: null } }, + { new: true }, + ) as IInstallableInstallation | null; + return throwIfLockLost(uninstalled); +}; + +const installAttempt = async ({ + installableId, + installedBy, + podId, +}: InstallArgs, recoveredStaleUninstall = false): Promise => { + const normalizedId = String(installableId || '').toLowerCase(); + const installable = await Installable.findOne({ + installableId: normalizedId, + status: 'active', + kind: 'app', + }) as IInstallable | null; + if (!installable) throw new InstallableNotFoundError(); + + const installerId = asObjectId(installedBy, 'installer'); + const targetPodId = asObjectId(podId, 'podId'); + const claim = await claimInstallation(installable, installerId, installerId, targetPodId); + if (!claim.ownsClaim) return resultForExisting(claim.installation, targetPodId); + + let installation = claim.installation; + const claimId = installation.claimId; + if (!claimId) throw new InstallLockLostError(); + + if (claim.claimedState === 'uninstalling') { + // A stale revocation has an intentionally terminal projection. Complete + // that teardown first, then claim a new parent so re-install never + // resurrects the old connector or its redeemed code. + await unprojectInstallation(installation, installable, installerId, claimId); + await finishUninstall(installation, claimId); + if (recoveredStaleUninstall) throw new InstallLockLostError(); + return installAttempt({ installableId, installedBy, podId }, true); + } + + if (claim.claimedState === 'installing') { + installation = await projectComponents( + installation, + installable, + installerId, + targetPodId, + claimId, + ); + installation = await transitionToActivating(installation, claimId); + } + + installation = await renewActivationLease(installation, claimId); + const integration = await activateIntegration(installation, targetPodId, claimId); + installation = await finishActivation(installation, claimId); + return { installation, integration, httpStatus: 200, state: 'active' }; +}; + +export const install = (args: InstallArgs): Promise => installAttempt(args); + +const claimUninstall = async ( + installation: IInstallableInstallation, +): Promise<{ installation: IInstallableInstallation; ownsClaim: boolean }> => { + const now = new Date(); + const claimId = randomUUID(); + const staleBefore = new Date(now.getTime() - INSTALL_LOCK_TTL_MS); + const isStaleTransient = ( + installation.status === 'installing' || installation.status === 'activating' + ) && isStale(installation.claimedAt, now); + if ( + (installation.status === 'installing' || installation.status === 'activating') + && !isStaleTransient + ) { + throw new InstallInProgressError(); + } + const filter = installation.status === 'uninstalling' + ? { + _id: installation._id, + status: 'uninstalling', + claimedAt: { $lte: staleBefore }, + } + : isStaleTransient + ? { + _id: installation._id, + status: { $in: ['installing', 'activating'] }, + claimedAt: { $lte: staleBefore }, + } + : { _id: installation._id, status: { $in: ['active', 'error'] } }; + const claimed = await InstallableInstallation.findOneAndUpdate( + filter, + { + $set: { + status: 'uninstalling', + claimId, + claimedAt: now, + errorMessage: null, + }, + }, + { new: true }, + ) as IInstallableInstallation | null; + if (claimed) return { installation: claimed, ownsClaim: true }; + + const winner = await InstallableInstallation.findById(installation._id) as IInstallableInstallation | null; + if (!winner || winner.status === 'uninstalled') throw new InstallableNotFoundError(); + if (winner.status === 'installing' || winner.status === 'activating') { + throw new InstallInProgressError(); + } + return { installation: winner, ownsClaim: false }; +}; + +export const uninstall = async ({ + installableId, + installedBy, +}: Pick): Promise => { + const targetId = asObjectId(installedBy, 'installer'); + const normalizedId = String(installableId || '').toLowerCase(); + const installation = await InstallableInstallation.findOne( + { ...installationKeys(normalizedId, targetId), status: { $ne: 'uninstalled' } }, + ) as IInstallableInstallation | null; + if (!installation) throw new InstallableNotFoundError(); + + const claim = await claimUninstall(installation); + if (!claim.ownsClaim) return claim.installation; + const claimId = claim.installation.claimId; + if (!claimId) throw new InstallLockLostError(); + + const installable = await Installable.findOne({ installableId: normalizedId }) as IInstallable | null; + await unprojectInstallation(claim.installation, installable, targetId, claimId); + return finishUninstall(claim.installation, claimId); +}; + +module.exports = { + INSTALL_LOCK_TTL_MS, + InstallLockLostError, + InstallableProjectionError, + InstallInProgressError, + InstallableAlreadyInstalledError, + InstallableNotFoundError, + install, + uninstall, +}; diff --git a/backend/services/installable/installableReconciler.ts b/backend/services/installable/installableReconciler.ts new file mode 100644 index 000000000..3a77dd675 --- /dev/null +++ b/backend/services/installable/installableReconciler.ts @@ -0,0 +1,137 @@ +import type { IInstallableInstallation } from '../../models/InstallableInstallation'; +import { INSTALL_LOCK_TTL_MS } from './installableInstallationService'; + +// eslint-disable-next-line @typescript-eslint/no-require-imports, global-require +const InstallableInstallation = require('../../models/InstallableInstallation'); +// eslint-disable-next-line @typescript-eslint/no-require-imports, global-require +const Integration = require('../../models/Integration'); + +const installationIdFor = (installation: IInstallableInstallation): string => String(installation._id); + +const sweepStaleLocks = async (now: Date): Promise<{ completed: number; errored: number }> => { + const staleBefore = new Date(now.getTime() - INSTALL_LOCK_TTL_MS); + const stale = await InstallableInstallation.find({ + status: { $in: ['installing', 'activating'] }, + claimedAt: { $lte: staleBefore }, + }).lean() as IInstallableInstallation[]; + let completed = 0; + let errored = 0; + + for (const installation of stale) { + const claimId = installation.claimId; + if (!claimId) continue; + if (installation.status === 'activating') { + const integration = await Integration.findOne({ + installationId: installationIdFor(installation), + }).lean() as { + isActive?: boolean; + config?: { connectCode?: string }; + } | null; + if ( + integration?.isActive + && integration.config?.connectCode + ) { + const result = await InstallableInstallation.updateOne( + { _id: installation._id, status: 'activating', claimId }, + { $set: { status: 'active', errorMessage: null } }, + ); + completed += result.modifiedCount || 0; + continue; + } + } + + const result = await InstallableInstallation.updateOne( + { _id: installation._id, status: installation.status, claimId }, + { $set: { status: 'error', errorMessage: 'install lock expired' } }, + ); + errored += result.modifiedCount || 0; + } + return { completed, errored }; +}; + +const sweepActiveInstallations = async (): Promise => { + const active = await InstallableInstallation.find({ status: 'active' }).lean() as IInstallableInstallation[]; + let staleComponents = 0; + for (const installation of active) { + const integration = await Integration.findOne({ installationId: installationIdFor(installation) }).lean(); + if (integration) continue; + const result = await InstallableInstallation.updateOne( + { _id: installation._id, status: 'active' }, + { $set: { 'components.$[].status': 'stale' } }, + ); + staleComponents += result.modifiedCount || 0; + } + return staleComponents; +}; + +const sweepUninstalledInstallations = async (): Promise => { + const rows = await InstallableInstallation.find({ status: 'uninstalled' }).lean() as IInstallableInstallation[]; + let deactivated = 0; + for (const installation of rows) { + const result = await Integration.updateOne( + { installationId: installationIdFor(installation), isActive: true }, + { + $set: { isActive: false }, + $unset: { + 'config.connectCode': 1, + 'config.connectCodeExpiresAt': 1, + }, + }, + ); + deactivated += result.modifiedCount || 0; + } + return deactivated; +}; + +const sweepStaleUninstalls = async (now: Date): Promise => { + const staleBefore = new Date(now.getTime() - INSTALL_LOCK_TTL_MS); + const rows = await InstallableInstallation.find({ + status: 'uninstalling', + claimedAt: { $lte: staleBefore }, + }).lean() as IInstallableInstallation[]; + let completed = 0; + + for (const installation of rows) { + if (!installation.claimId) continue; + // This is the revocation equivalent of the activating split-commit + // recovery. Parent state is only finalized after the projection is made + // inactive; the claim fence makes a stale sweep harmless to a takeover. + await Integration.updateOne( + { installationId: installationIdFor(installation) }, + { + $set: { + isActive: false, + revokedAt: new Date(), + }, + $unset: { + 'config.connectCode': 1, + 'config.connectCodeExpiresAt': 1, + }, + }, + ); + const result = await InstallableInstallation.updateOne( + { _id: installation._id, status: 'uninstalling', claimId: installation.claimId }, + { $set: { status: 'uninstalled', errorMessage: null } }, + ); + completed += result.modifiedCount || 0; + } + return completed; +}; + +export const sweep = async (now: Date = new Date()): Promise<{ + completed: number; + errored: number; + staleComponents: number; + uninstallsCompleted: number; + deactivated: number; +}> => { + const locks = await sweepStaleLocks(now); + const staleComponents = await sweepActiveInstallations(); + const uninstallsCompleted = await sweepStaleUninstalls(now); + const deactivated = await sweepUninstalledInstallations(); + const result = { ...locks, staleComponents, uninstallsCompleted, deactivated }; + console.log('[installable-reconciler] sweep', result); + return result; +}; + +module.exports = { sweep }; diff --git a/backend/services/installable/projectors/eventHandlerProjector.ts b/backend/services/installable/projectors/eventHandlerProjector.ts new file mode 100644 index 000000000..73d3fe863 --- /dev/null +++ b/backend/services/installable/projectors/eventHandlerProjector.ts @@ -0,0 +1,36 @@ +import type { IComponent } from '../../../models/Installable'; +import type { ComponentProjector, ProjectionContext, ProjectionIds } from './types'; + +// eslint-disable-next-line @typescript-eslint/no-require-imports, global-require +const Integration = require('../../../models/Integration'); +// eslint-disable-next-line @typescript-eslint/no-require-imports, global-require +const { hasEventHandler } = require('../eventHandlers'); + +export const eventHandlerProjector: ComponentProjector = { + type: 'event-handler', + + async project(component: IComponent, context: ProjectionContext): Promise { + const handler = component.eventHandler; + if (!handler || !hasEventHandler(handler)) { + throw new Error(`Unknown internal event handler: ${handler || '(missing)'}`); + } + + const integration = await Integration.findOne({ + installationId: String(context.installation._id), + }); + if (!integration) { + throw new Error('Event-handler projection requires its webhook Integration row'); + } + return { integrationId: integration._id }; + }, + + async unproject( + _component: IComponent, + _context: ProjectionContext, + _projectionIds: ProjectionIds, + ): Promise { + // The webhook projector owns the shared Integration lifecycle. Removing an + // event handler only removes its parent component record; it must not + // deactivate a row another component of the same installation still owns. + }, +}; diff --git a/backend/services/installable/projectors/index.ts b/backend/services/installable/projectors/index.ts new file mode 100644 index 000000000..5517adbcd --- /dev/null +++ b/backend/services/installable/projectors/index.ts @@ -0,0 +1,15 @@ +import { eventHandlerProjector } from './eventHandlerProjector'; +import type { ComponentProjector } from './types'; +import { webhookProjector } from './webhookProjector'; + +export const projectorRegistry: Map = new Map([ + [webhookProjector.type, webhookProjector], + [eventHandlerProjector.type, eventHandlerProjector], +]); + +export const getProjector = (type: string): ComponentProjector | undefined => projectorRegistry.get(type); + +module.exports = { + projectorRegistry, + getProjector, +}; diff --git a/backend/services/installable/projectors/types.ts b/backend/services/installable/projectors/types.ts new file mode 100644 index 000000000..9c1c6c48f --- /dev/null +++ b/backend/services/installable/projectors/types.ts @@ -0,0 +1,26 @@ +import type { Types } from 'mongoose'; + +import type { IComponent, IInstallable } from '../../../models/Installable'; +import type { IInstallableInstallation } from '../../../models/InstallableInstallation'; + +export type ProjectionIds = Record; + +export interface ProjectionContext { + installation: IInstallableInstallation; + installable: IInstallable; + installedBy: Types.ObjectId; + // A pod is required while creating a pod-scoped projection. Unprojection + // acts on the recorded projection ID and must not invent a target pod. + podId?: Types.ObjectId; + claimId: string; +} + +export interface ComponentProjector { + type: IComponent['type']; + project(component: IComponent, context: ProjectionContext): Promise; + unproject( + component: IComponent, + context: ProjectionContext, + projectionIds: ProjectionIds, + ): Promise; +} diff --git a/backend/services/installable/projectors/webhookProjector.ts b/backend/services/installable/projectors/webhookProjector.ts new file mode 100644 index 000000000..73a029f8a --- /dev/null +++ b/backend/services/installable/projectors/webhookProjector.ts @@ -0,0 +1,78 @@ +import type { IComponent } from '../../../models/Installable'; +import type { ComponentProjector, ProjectionContext, ProjectionIds } from './types'; + +// eslint-disable-next-line @typescript-eslint/no-require-imports, global-require +const Integration = require('../../../models/Integration'); + +const INTERNAL_WEBHOOK_PROVIDERS: Record = { + '/api/webhooks/telegram': 'telegram', +}; + +const installationIdFor = (context: ProjectionContext): string => String(context.installation._id); + +const resolveProvider = (component: IComponent): 'telegram' => { + const provider = component.webhookPath + ? INTERNAL_WEBHOOK_PROVIDERS[component.webhookPath] + : undefined; + if (!provider) { + throw new Error(`Unsupported internal webhook path: ${component.webhookPath || '(missing)'}`); + } + return provider; +}; + +export const webhookProjector: ComponentProjector = { + type: 'webhook', + + async project(component: IComponent, context: ProjectionContext): Promise { + const provider = resolveProvider(component); + const installationId = installationIdFor(context); + if (!context.podId) { + throw new Error('Webhook projection requires a pod target'); + } + + // Projection never mints a code. A partially projected row must remain + // impossible for the unauthenticated enable route to redeem. + const integration = await Integration.findOneAndUpdate( + { installationId }, + { + $setOnInsert: { + installationId, + podId: context.podId, + type: provider, + status: 'pending', + createdBy: context.installedBy, + isActive: false, + config: { + liveRelay: true, + relayAllAgentMessages: true, + linkedUserId: String(context.installedBy), + }, + }, + }, + { new: true, upsert: true, setDefaultsOnInsert: true }, + ); + + if (!integration) throw new Error('Webhook projection did not return an Integration row'); + return { integrationId: integration._id }; + }, + + async unproject( + _component: IComponent, + context: ProjectionContext, + _projectionIds: ProjectionIds, + ): Promise { + await Integration.findOneAndUpdate( + { installationId: installationIdFor(context) }, + { + $set: { + isActive: false, + revokedAt: new Date(), + }, + $unset: { + 'config.connectCode': 1, + 'config.connectCodeExpiresAt': 1, + }, + }, + ); + }, +}; diff --git a/backend/services/telegramBridgeService.ts b/backend/services/telegramBridgeService.ts index 2ad26a776..2f0c7167e 100644 --- a/backend/services/telegramBridgeService.ts +++ b/backend/services/telegramBridgeService.ts @@ -35,6 +35,8 @@ export interface RelayMapEntry { interface TelegramIntegrationDoc { _id: unknown; podId: unknown; + type?: string; + isActive?: boolean; config?: { chatId?: string; chatType?: string; @@ -123,6 +125,18 @@ const findLiveIntegration = async (podId: unknown): Promise ( + String(integration.podId) === String(podId) + && (integration.type === undefined || integration.type === 'telegram') + && integration.isActive !== false + && integration.config?.liveRelay === true + && integration.config?.chatType === 'private' + && Boolean(integration.config?.chatId) +); + // Outbound: agent message → Telegram, attributed, with a deep link back into // the pod. Fire-and-forget from AgentMessageService.postMessage — a bridge // failure must never fail the post itself. @@ -132,13 +146,16 @@ export const relayAgentMessageToTelegram = async (opts: { displayName: string; content: string; podMessageId?: string | null; + integration?: TelegramIntegrationDoc; }): Promise => { const { podId, agentUsername, displayName, content, podMessageId, } = opts; try { - const integration = await findLiveIntegration(podId); - if (!integration) return; + // The dispatcher selects each pod-scoped subscription. Its row is the + // authority for this send; the fallback preserves legacy direct rows. + const integration = opts.integration ?? await findLiveIntegration(podId); + if (!integration || !isRelayableIntegration(integration, podId)) return; // /mute pauses ALL outbound relay to the chat, escalations included — // mute means mute; /status shows it, and it self-expires. const mutedUntil = (integration.config as { relayMutedUntil?: Date | string })?.relayMutedUntil; @@ -234,7 +251,8 @@ export const relayTelegramMessageToPod = async (opts: { const chatType = integration.config?.chatType; if (chatType !== 'private') { console.warn( - `[tg-bridge] inbound dropped — relay authors as the linked user and chatType=${chatType || 'unknown'} cannot guarantee the sender is them`, + `[tg-bridge] inbound dropped — relay authors as the linked user and chatType=${chatType || 'unknown'} ` + + 'cannot guarantee the sender is them', ); return { relayed: false }; } diff --git a/docs/plans/connector-as-installable-app.md b/docs/plans/connector-as-installable-app.md index 5133bc838..6298850d5 100644 --- a/docs/plans/connector-as-installable-app.md +++ b/docs/plans/connector-as-installable-app.md @@ -90,12 +90,14 @@ What it does, in order — this is `installableInstallService.install()`: 3. **Claim the parent atomically — the claim is the compare-and-set** (the #1315 shape; Kai's ask, 2026-09-02). One row per `(installableId, targetType, targetId)` across every non-uninstalled state: a **unique partial index filtered to `status: { $in: ['installing', - 'activating', 'active', 'error'] }`**. The service never reads-then-writes. It runs one + 'activating', 'uninstalling', 'active', 'error'] }`**. The service never reads-then-writes. It runs one `findOneAndUpdate` with `upsert: true` whose filter is the key plus **one of**: no row; - `status: 'error'`; or `status: { $in: ['installing', 'activating'] }` with `claimedAt < - now − INSTALL_LOCK_TTL_MS` - — and whose update sets `status: 'installing'`, **`claimId: randomUUID()` — a fresh - generation on every claim and every takeover**, `installedBy`, `installableVersion`, + `status: 'error'`; `status: { $in: ['installing', 'activating'] }` with `claimedAt < + now − INSTALL_LOCK_TTL_MS`; or stale `status: 'uninstalling'` with the same predicate. + — and whose update claims `installing` for a new/error parent, retains `activating` to resume + its split write, and retains `uninstalling` until its terminal teardown completes; every path + writes **`claimId: randomUUID()` — a fresh generation on every claim and every takeover**, + `installedBy`, `installableVersion`, `installSource: 'ui'`, `grantedScopes = installable.requires` (**descriptive only in Phase 1** — it records what the manifest declared at install time, mirroring ADR-001's "declared, permissive enforcement"; nothing reads it for authorization, and no route may start to @@ -153,9 +155,11 @@ What it does, in order — this is `installableInstallService.install()`: applies), and **the idempotent-return path never treats it as success** — a retry or a takeover that finds `activating` skips the projectors and resumes at write 2. 2. **Integration activation, fenced on its own state:** `findOneAndUpdate({ installationId: - String(parent._id), isActive: false }, { $set: { isActive: true, 'config.connectCode': - …, 'config.connectCodeExpiresAt': … } })` — `mintConnectCode()` is called exactly once, - inside this write's construction. `null` here means the row is already active (a + String(parent._id), isActive: false, revokedAt: { $exists: false } }, { $set: { + isActive: true, installationClaimId: ours, 'config.connectCode': …, + 'config.connectCodeExpiresAt': … } })` — `mintConnectCode()` is called exactly once, + inside this write's construction. `installationClaimId` records the generation that + minted the code; it is not an eligibility predicate. `null` here means the row is already active (a resume after a crash between writes 2 and 3): read the existing code, mint nothing. 3. **Parent CAS → `active`:** `findOneAndUpdate({ _id, status: 'activating', claimId: ours }, { $set: { status: 'active' } })`, fenced like every other owner write; `null` ⇒ @@ -218,11 +222,16 @@ independently of this spec. **Nothing here writes `installationId` until that te 'user'`, `targetId: req.user.id` — and from nothing else: the route takes no installation id and no body field, so there is no way to name someone else's row. (Vera, 2026-09-02: install gated the pod by `isPodMember` while uninstall had no matching gate — a co-member could have -torn down another member's connector.) The live row for that key (`installing` or `active`) -goes → `uninstalled`; each projector's `unproject` runs (Integration `isActive: false`, connect -code cleared, relayMap kept for audit); nothing is deleted; no row → 404. Uninstalling an -`installing` row is allowed and is the human escape hatch for a stuck lock in addition to the -lease. Re-install mints a new Integration row — Vera's +torn down another member's connector.) Revocation is a recoverable split commit: the parent +first becomes **`uninstalling`** under a fresh generation; projectors deactivate their rows and +clear the code; only then does a generation-fenced CAS finalize it as `uninstalled`. The webhook +projection writes a terminal `revokedAt` tombstone before finalization, so an old activation +generation cannot revive a connector after revocation. A fresh concurrent delete gets 202 +`uninstalling`, never a false disconnected success; the reconciler deactivates a stale +`uninstalling` projection before it completes the parent. Nothing is deleted; no row → 404. +Uninstalling an `installing` row is allowed and is the human escape hatch for a stuck lock in +addition to the lease. Install also takes over a stale `uninstalling` row, completes its terminal +teardown, and then claims a new parent. Re-install mints a new Integration row — Vera's ruling on the design spec stands (the binding row is the unit; relayMap and gates are never reused). @@ -273,18 +282,22 @@ and the promise moves up a layer with the call). The dispatcher then invokes eac handler with the same payload the bridge takes today (`{ podId, agentUsername, displayName, content, podMessageId }`) plus the selected `integration`, fire-and-forget, one `try/catch` per handler so one bridge cannot fail the post. The bridge's own `findLiveIntegration(podId)` stays -in Phase 1 as defence in depth, not as the selector: an install must be filtered out **before** -its handler runs, not inside it — a dispatcher that fans out to every tenant and relies on each -handler to decline is a multi-tenant leak waiting for a handler that does not (Vera, -2026-09-02). In Phase 2 the selector becomes D8's inversion — pod → members → each member's -user-scoped install — and the bridge lookup is deleted; the handler signature does not change. +in Phase 1 only as a fallback for legacy direct rows. For a dispatched handler, the bridge +honours the selected `integration` and independently validates its provider, pod, active, +live-relay, private-chat, and chat-id gates; it does not re-select with `findOne`. An install +must be filtered out **before** its handler runs — a dispatcher that fans out to every tenant +and relies on each handler to decline is a multi-tenant leak waiting for a handler that does not +(Vera, 2026-09-02). In Phase 2 the selector becomes D8's inversion — pod → members → each +member's user-scoped install — and the legacy fallback is deleted; the handler signature does +not change. **Behaviour pins:** (a) for a pod with one live Telegram row, exactly one relay fires per post, with the same arguments as before; (b) **two tenants**: user A's install bound to pod P and user B's bound to pod Q — a post in P invokes A's handler once and B's zero times, measured at the dispatcher (a spy on the handler map), not at the bridge; (c) a pod with no install costs -one selector query and zero invocations. Those are the tests that prove invariant 6 landed -without moving the product or widening it. +one selector query and zero invocations; (d) two users' active installs bound to the same pod +are two subscriptions, so one post produces one send to each selected private chat. Those are +the tests that prove invariant 6 landed without moving the product or widening it. Unknown `eventHandler` prefix (`agent:`, `webhook:`) → projector error in Phase 1. Those are the slash-command / external-webhook tracks; naming them here keeps the enum honest. @@ -399,6 +412,11 @@ Unit (`backend/__tests__/unit/services/installable/`): calls `DELETE /api/installables/telegram/install`: A's row is untouched (B gets 404 with no install of their own, or uninstalls only their own). A body containing another installation's id changes nothing. +4c. **Revocation never reports success ahead of its projection.** A projector failure or crash + after the parent enters `uninstalling` leaves the parent recoverable, not `uninstalled`; the + Integration becomes inactive with no code before a sweep or retry can finalize it. A stale + sweep deactivates first and then fences the `uninstalling → uninstalled` transition; the + terminal tombstone prevents an old activation generation from reviving the Integration. 5. Non-member of the chosen pod → 403, nothing written. 6. Reconciler: a deleted Integration under an active installation marks the component `stale`, creates nothing. An `installing` row with `claimedAt` older than the TTL is swept to @@ -412,8 +430,7 @@ Unit (`backend/__tests__/unit/services/installable/`): `installing` row (within the TTL) is not taken over — the second caller gets 202. 6c. **Stale-owner completion is a refused no-op, and it is visible.** A claims (generation `a`) and stalls; B takes over (generation `b`), activates, mints `C_B`. A revives and runs - its activation: write 1 returns `null`, `InstallLockLostError` is thrown, **no mint call - happens** (spy on `mintConnectCode`: exactly one call in the whole test, B's), **no + its activation is refused with `InstallLockLostError`, **no second code is stored**, **no `unproject` call happens and A writes nothing** (spies on the projector registry and on the `Integration` model: B's writes only), the Integration row still carries `C_B` and stays `isActive: true`, and A's caller receives 409 `install_lock_lost` — asserted on the status @@ -434,9 +451,9 @@ Service (`__tests__/service/`): row triggers exactly one `relayAgentMessageToTelegram` call via the dispatcher, with the same five fields the hardcoded hook passed. (b) **Multi-tenant:** two active installs — user A's bound to pod P, user B's bound to pod Q — and a post in P: A's handler is invoked once, B's - zero times, asserted on a spy at the handler map (the bridge's own lookup is stubbed out so - it cannot be what filtered B). (c) A pod with no install: one selector query, zero - invocations. (d) A throwing handler does not fail the post. + zero times, asserted on a spy at the handler map. (c) A pod with no install: one selector + query, zero invocations. (d) Two active user installs on the same pod produce exactly two + real Telegram sends, one to each selected chat. (e) A throwing handler does not fail the post. 8. `telegramBridgeService.attribution` and `telegram.webhook.*` suites pass unchanged — the route is not edited.