Nick SzaunigandClaude Opus 5.5
8a31c42578
- Migrations (additive): veterinary medicines with lots and movement ledger, treatment medicine/lot/dose links, label particulars, authority notices, apiary restrictions, compliance settings and acknowledgements; notification category COMPLIANCE, added to non-empty Telegram lists. - Permissions veterinary-medicines:* and compliance:* with role grants. - LOCKABLE_TABLES: Apiary locks before Colony; T6 tables appended. - Backup exports and restores the new tables in FK order; unsealed files skip them and report dropped medicine links; an open TSK draft in the target wins. - describeNullContracts shared helper; Telegram backlog e2e no longer depends on the time of day. Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
1073 lines
38 KiB
TypeScript
1073 lines
38 KiB
TypeScript
import { createHash } from 'node:crypto';
|
|
import { INestApplication } from '@nestjs/common';
|
|
import * as bcrypt from 'bcryptjs';
|
|
import request from 'supertest';
|
|
import { App } from 'supertest/types';
|
|
import type { NotificationCategory } from '../generated/prisma';
|
|
import { berlinDateTime, berlinToday, dayToDate } from '../src/common/dates';
|
|
import { DomainEventsService } from '../src/domain-events/domain-events.service';
|
|
import { PermissionResolverService } from '../src/auth/permissions/permission-resolver.service';
|
|
import {
|
|
DeliveryPreferences,
|
|
isInQuietHours,
|
|
} from '../src/notifications/delivery-time';
|
|
import { NotificationsService } from '../src/notifications/notifications.service';
|
|
import { PrismaService } from '../src/prisma/prisma.service';
|
|
import { TelegramClient } from '../src/telegram/telegram-client';
|
|
import { TelegramLinkService } from '../src/telegram/telegram-link.service';
|
|
import { TelegramSender } from '../src/telegram/telegram-sender';
|
|
import { TelegramUpdatesService } from '../src/telegram/telegram-updates.service';
|
|
import { login } from './utils/api';
|
|
import { drainDomainEvents } from './utils/domain-events';
|
|
import { createTelegramTestApp, FakeTelegramClient } from './utils/telegram';
|
|
|
|
const HOUR_MS = 3_600_000;
|
|
const CLOCK_TOLERANCE_MS = 60_000;
|
|
const ALL_CATEGORIES: NotificationCategory[] = [
|
|
'TASK_DUE',
|
|
'DIGEST',
|
|
'SUGGESTION',
|
|
'SYSTEM',
|
|
];
|
|
const NEVER_ABORTED = new AbortController().signal;
|
|
|
|
describe('Telegram delivery (e2e)', () => {
|
|
let app: INestApplication<App>;
|
|
let prisma: PrismaService;
|
|
let telegram: FakeTelegramClient;
|
|
let sender: TelegramSender;
|
|
const suffix = `${Date.now()}-${Math.round(Math.random() * 1e6)}`;
|
|
const chatBase = 8_000_000_000n + BigInt(Math.round(Math.random() * 1e6));
|
|
let chatCounter = 0n;
|
|
const roleIds: string[] = [];
|
|
const userIds: string[] = [];
|
|
const taskIds: string[] = [];
|
|
|
|
interface TestUser {
|
|
id: string;
|
|
chatId: bigint;
|
|
}
|
|
|
|
async function createUser(
|
|
label: string,
|
|
permissions: string[],
|
|
preference: {
|
|
telegramCategories?: NotificationCategory[];
|
|
quietFrom?: string | null;
|
|
quietUntil?: string | null;
|
|
digestTime?: string;
|
|
telegramDetail?: boolean;
|
|
} = {},
|
|
linked = true,
|
|
): Promise<TestUser & { roleId: string }> {
|
|
const role = await prisma.role.create({
|
|
data: {
|
|
key: `TELEGRAM_${label.toUpperCase()}_${suffix}`,
|
|
name: `Telegram ${label}`,
|
|
permissions: {
|
|
create: permissions.map((key) => ({
|
|
permission: { connect: { key } },
|
|
})),
|
|
},
|
|
},
|
|
});
|
|
roleIds.push(role.id);
|
|
const user = await prisma.user.create({
|
|
data: {
|
|
email: `telegram-e2e-${label}-${suffix}@example.com`,
|
|
passwordHash: await bcrypt.hash('SuperSecret123!', 4),
|
|
name: `Telegram ${label}`,
|
|
role: { connect: { id: role.id } },
|
|
},
|
|
});
|
|
userIds.push(user.id);
|
|
const chatId = chatBase + ++chatCounter;
|
|
await prisma.notificationPreference.create({
|
|
data: {
|
|
userId: user.id,
|
|
telegramEnabled: true,
|
|
telegramCategories: ALL_CATEGORIES,
|
|
...preference,
|
|
},
|
|
});
|
|
if (linked) {
|
|
await prisma.telegramLink.create({
|
|
data: { userId: user.id, chatId, telegramFirstName: label },
|
|
});
|
|
}
|
|
return { id: user.id, chatId, roleId: role.id };
|
|
}
|
|
|
|
/** A notification with its pending delivery, as the planner leaves them. */
|
|
async function queue(
|
|
userId: string,
|
|
options: {
|
|
category?: NotificationCategory;
|
|
message?: string;
|
|
requiredPermissions?: string[];
|
|
createdAgoMs?: number;
|
|
relatedType?: string;
|
|
relatedId?: string;
|
|
link?: string;
|
|
} = {},
|
|
): Promise<{ notificationId: string; deliveryId: string }> {
|
|
const notification = await prisma.notification.create({
|
|
data: {
|
|
userId,
|
|
priority: 'INFO',
|
|
category: options.category ?? 'SYSTEM',
|
|
message: options.message ?? 'Hallo vom Test',
|
|
requiredPermissions: options.requiredPermissions ?? [],
|
|
relatedType: options.relatedType ?? null,
|
|
relatedId: options.relatedId ?? null,
|
|
link: options.link ?? '/dashboard',
|
|
createdAt: new Date(Date.now() - (options.createdAgoMs ?? 0)),
|
|
},
|
|
});
|
|
const delivery = await prisma.notificationDelivery.create({
|
|
data: {
|
|
channel: 'TELEGRAM',
|
|
userId,
|
|
notificationId: notification.id,
|
|
availableAt: new Date(Date.now() - 1_000),
|
|
},
|
|
});
|
|
return { notificationId: notification.id, deliveryId: delivery.id };
|
|
}
|
|
|
|
const drainFor = (...ids: string[]) =>
|
|
sender.drain(NEVER_ABORTED, 5, { userIds: ids });
|
|
const deliveryOf = (id: string) =>
|
|
prisma.notificationDelivery.findUniqueOrThrow({ where: { id } });
|
|
|
|
beforeAll(async () => {
|
|
({ app, telegram } = await createTelegramTestApp());
|
|
prisma = app.get(PrismaService);
|
|
sender = app.get(TelegramSender);
|
|
});
|
|
|
|
afterAll(async () => {
|
|
await prisma.notificationDelivery.deleteMany({
|
|
where: { userId: { in: userIds } },
|
|
});
|
|
await prisma.user.deleteMany({ where: { id: { in: userIds } } });
|
|
await prisma.task.deleteMany({ where: { id: { in: taskIds } } });
|
|
await prisma.role.deleteMany({ where: { id: { in: roleIds } } });
|
|
await app.close();
|
|
});
|
|
|
|
describe('planning', () => {
|
|
it('a due task yields one delivery after the morning time and none inside quiet hours', async () => {
|
|
const today = berlinToday();
|
|
const morning: DeliveryPreferences = {
|
|
digestTime: '12:00',
|
|
quietFrom: null,
|
|
quietUntil: null,
|
|
};
|
|
const night: DeliveryPreferences = {
|
|
digestTime: '06:30',
|
|
quietFrom: '00:00',
|
|
quietUntil: '23:59',
|
|
};
|
|
const early = await createUser('plan-early', ['tasks:read'], morning);
|
|
const quiet = await createUser('plan-quiet', ['tasks:read'], night);
|
|
const off = await createUser('plan-off', ['tasks:read'], {
|
|
telegramCategories: ['DIGEST'],
|
|
});
|
|
const task = await prisma.task.create({
|
|
data: {
|
|
title: `Telegram e2e ${suffix}`,
|
|
dueDate: dayToDate(today),
|
|
},
|
|
});
|
|
taskIds.push(task.id);
|
|
// Without an assignee every active user with tasks:complete gets it;
|
|
// assign it to each user in turn to keep the test to its own users.
|
|
const before = new Date();
|
|
for (const user of [early, quiet, off]) {
|
|
await prisma.task.update({
|
|
where: { id: task.id },
|
|
data: { assignedUserId: user.id },
|
|
});
|
|
await prisma.$transaction((tx) =>
|
|
app.get(DomainEventsService).record(tx, {
|
|
type: 'TaskDue',
|
|
aggregateType: 'Task',
|
|
aggregateId: task.id,
|
|
payload: { dueOn: today },
|
|
dedupeKey: `TaskDue:${task.id}:${today}:${user.id}`,
|
|
}),
|
|
);
|
|
await drainDomainEvents(app, [task.id]);
|
|
}
|
|
|
|
const forUser = (userId: string) =>
|
|
prisma.notificationDelivery.findMany({
|
|
where: { userId },
|
|
include: { notification: true },
|
|
});
|
|
const [earlyRow, ...earlyRest] = await forUser(early.id);
|
|
const [quietRow, ...quietRest] = await forUser(quiet.id);
|
|
|
|
expect(earlyRow.notification?.category).toBe('TASK_DUE');
|
|
expect(earlyRest).toHaveLength(0);
|
|
expect(earlyRow.channel).toBe('TELEGRAM');
|
|
expect(earlyRow.status).toBe('PENDING');
|
|
// Not before one minute after the morning time of that day; when that
|
|
// time has passed already, right away.
|
|
const morningAt = berlinDateTime(today, '12:01');
|
|
if (before < morningAt) {
|
|
expect(earlyRow.availableAt).toEqual(morningAt);
|
|
} else {
|
|
expect(earlyRow.availableAt.getTime()).toBeGreaterThanOrEqual(
|
|
before.getTime() - CLOCK_TOLERANCE_MS,
|
|
);
|
|
}
|
|
expect(isInQuietHours(earlyRow.availableAt, morning)).toBe(false);
|
|
// Quiet hours all day: the delivery waits for their end.
|
|
expect(quietRest).toHaveLength(0);
|
|
expect(isInQuietHours(quietRow.availableAt, night)).toBe(false);
|
|
expect(quietRow.availableAt.getTime()).toBeGreaterThanOrEqual(
|
|
before.getTime() - CLOCK_TOLERANCE_MS,
|
|
);
|
|
// The user who did not choose TASK_DUE has the notification, no delivery.
|
|
expect(await forUser(off.id)).toHaveLength(0);
|
|
expect(
|
|
await prisma.notification.count({
|
|
where: { userId: off.id, category: 'TASK_DUE' },
|
|
}),
|
|
).toBe(1);
|
|
});
|
|
|
|
it('a user without a link gets no delivery', async () => {
|
|
const user = await createUser('plan-unlinked', [], {}, false);
|
|
|
|
await app.get(NotificationsService).record({
|
|
userId: user.id,
|
|
priority: 'INFO',
|
|
category: 'SYSTEM',
|
|
message: 'm',
|
|
});
|
|
|
|
expect(
|
|
await prisma.notificationDelivery.count({ where: { userId: user.id } }),
|
|
).toBe(0);
|
|
});
|
|
|
|
it('queues the SYSTEM message of a user who chose that category, once, even if the key repeats', async () => {
|
|
const user = await createUser('plan-system', [], {
|
|
telegramCategories: ['SYSTEM'],
|
|
quietFrom: null,
|
|
quietUntil: null,
|
|
});
|
|
const input = {
|
|
userId: user.id,
|
|
priority: 'INFO' as const,
|
|
category: 'SYSTEM' as const,
|
|
message: 'm',
|
|
dedupeKey: `telegram-e2e-system:${suffix}`,
|
|
};
|
|
|
|
await app.get(NotificationsService).record(input);
|
|
await app.get(NotificationsService).record(input);
|
|
|
|
expect(
|
|
await prisma.notificationDelivery.count({ where: { userId: user.id } }),
|
|
).toBe(1);
|
|
});
|
|
});
|
|
|
|
describe('sending', () => {
|
|
it('two send loops in parallel deliver each message exactly once', async () => {
|
|
const users = await Promise.all(
|
|
['par-a', 'par-b', 'par-c', 'par-d', 'par-e'].map((label) =>
|
|
createUser(label, []),
|
|
),
|
|
);
|
|
const queued = await Promise.all(
|
|
users.map((user) =>
|
|
queue(user.id, { message: `Nachricht ${user.id}` }),
|
|
),
|
|
);
|
|
const second = new TelegramSender(
|
|
prisma,
|
|
telegram as never,
|
|
app.get(PermissionResolverService),
|
|
app.get(NotificationsService),
|
|
);
|
|
const ids = users.map((user) => user.id);
|
|
|
|
const claimed = await Promise.all([
|
|
sender.drain(NEVER_ABORTED, 5, { userIds: ids }),
|
|
second.drain(NEVER_ABORTED, 5, { userIds: ids }),
|
|
]);
|
|
|
|
expect(claimed[0] + claimed[1]).toBe(users.length);
|
|
for (const user of users) {
|
|
expect(telegram.sentTo(user.chatId)).toHaveLength(1);
|
|
}
|
|
for (const { deliveryId } of queued) {
|
|
const row = await deliveryOf(deliveryId);
|
|
expect(row.status).toBe('DELIVERED');
|
|
expect(row.deliveredAt).not.toBeNull();
|
|
}
|
|
});
|
|
|
|
it('sends plain text with the app link', async () => {
|
|
const user = await createUser('text', [], {});
|
|
const { deliveryId } = await queue(user.id, {
|
|
message: 'Heute fällig: Futter',
|
|
link: '/tasks',
|
|
});
|
|
|
|
await drainFor(user.id);
|
|
|
|
const [message] = telegram.sentTo(user.chatId);
|
|
expect(message.text).toMatch(
|
|
/^Heute fällig: Futter\n\nhttps?:\/\/.+\/tasks$/,
|
|
);
|
|
expect((await deliveryOf(deliveryId)).status).toBe('DELIVERED');
|
|
});
|
|
|
|
it('a revoked permission skips the delivery at send time', async () => {
|
|
const user = await createUser('revoked', ['tasks:read']);
|
|
const { deliveryId } = await queue(user.id, {
|
|
requiredPermissions: ['tasks:read'],
|
|
});
|
|
// The resolver in this process has the right cached ...
|
|
await app.get(PermissionResolverService).getPermissions(user.id);
|
|
// ... and loses it without any invalidateUser call.
|
|
await prisma.rolePermission.deleteMany({
|
|
where: { roleId: user.roleId },
|
|
});
|
|
|
|
await drainFor(user.id);
|
|
|
|
expect(telegram.sentTo(user.chatId)).toHaveLength(0);
|
|
const row = await deliveryOf(deliveryId);
|
|
expect(row.status).toBe('SKIPPED');
|
|
expect(row.lastError).toBe('permission revoked');
|
|
});
|
|
|
|
it('skips what went stale: a DIGEST of yesterday, a SUGGESTION older than 24 h', async () => {
|
|
const user = await createUser('stale', [], {
|
|
telegramCategories: ['DIGEST', 'SUGGESTION', 'SYSTEM'],
|
|
});
|
|
const digest = await queue(user.id, {
|
|
category: 'DIGEST',
|
|
createdAgoMs: 36 * HOUR_MS,
|
|
});
|
|
const old = await queue(user.id, {
|
|
category: 'SUGGESTION',
|
|
createdAgoMs: 25 * HOUR_MS,
|
|
});
|
|
await drainFor(user.id);
|
|
const fresh = await queue(user.id, {
|
|
category: 'SUGGESTION',
|
|
createdAgoMs: 23 * HOUR_MS,
|
|
});
|
|
|
|
await drainFor(user.id);
|
|
|
|
expect((await deliveryOf(digest.deliveryId)).lastError).toBe('stale');
|
|
expect((await deliveryOf(old.deliveryId)).status).toBe('SKIPPED');
|
|
expect((await deliveryOf(fresh.deliveryId)).status).toBe('DELIVERED');
|
|
expect(telegram.sentTo(user.chatId)).toHaveLength(1);
|
|
});
|
|
|
|
it('skips a due-task reminder once the task is done', async () => {
|
|
const user = await createUser('taskdone', ['tasks:read'], {
|
|
telegramCategories: ['TASK_DUE'],
|
|
});
|
|
const task = await prisma.task.create({
|
|
data: { title: `Done ${suffix}`, status: 'DONE' },
|
|
});
|
|
taskIds.push(task.id);
|
|
const { deliveryId } = await queue(user.id, {
|
|
category: 'TASK_DUE',
|
|
relatedType: 'Task',
|
|
relatedId: task.id,
|
|
requiredPermissions: ['tasks:read'],
|
|
});
|
|
|
|
await drainFor(user.id);
|
|
|
|
expect(telegram.sentTo(user.chatId)).toHaveLength(0);
|
|
expect((await deliveryOf(deliveryId)).lastError).toBe(
|
|
'task already done',
|
|
);
|
|
});
|
|
|
|
it('skips when Telegram was switched off, the link is gone or the category is off', async () => {
|
|
const off = await createUser('sw-off', [], {});
|
|
await prisma.notificationPreference.update({
|
|
where: { userId: off.id },
|
|
data: { telegramEnabled: false },
|
|
});
|
|
const unlinked = await createUser('sw-unlinked', [], {}, false);
|
|
const category = await createUser('sw-category', [], {
|
|
telegramCategories: ['DIGEST'],
|
|
});
|
|
const rows = await Promise.all([
|
|
queue(off.id),
|
|
queue(unlinked.id),
|
|
queue(category.id),
|
|
]);
|
|
|
|
await drainFor(off.id, unlinked.id, category.id);
|
|
|
|
const reasons = await Promise.all(
|
|
rows.map(async (row) => (await deliveryOf(row.deliveryId)).lastError),
|
|
);
|
|
expect(reasons).toEqual([
|
|
'telegram disabled',
|
|
'not linked',
|
|
'category off',
|
|
]);
|
|
});
|
|
|
|
it('403 removes the link, switches Telegram off and notifies the user in the app', async () => {
|
|
const user = await createUser('blocked', []);
|
|
const { deliveryId } = await queue(user.id);
|
|
telegram.failWith(
|
|
403,
|
|
'Forbidden: bot was blocked by the user',
|
|
user.chatId,
|
|
);
|
|
|
|
await drainFor(user.id);
|
|
|
|
expect(
|
|
await prisma.telegramLink.count({ where: { userId: user.id } }),
|
|
).toBe(0);
|
|
const preference = await prisma.notificationPreference.findUniqueOrThrow({
|
|
where: { userId: user.id },
|
|
});
|
|
expect(preference.telegramEnabled).toBe(false);
|
|
const row = await deliveryOf(deliveryId);
|
|
expect(row.status).toBe('SKIPPED');
|
|
expect(row.lastError).toBe('chat unavailable');
|
|
const notices = await prisma.notification.findMany({
|
|
where: {
|
|
userId: user.id,
|
|
category: 'SYSTEM',
|
|
message: { contains: 'Zustellung beendet' },
|
|
},
|
|
});
|
|
expect(notices).toHaveLength(1);
|
|
});
|
|
|
|
it('5xx keeps the delivery for a retry with the attempt counted', async () => {
|
|
const user = await createUser('server-error', []);
|
|
const { deliveryId } = await queue(user.id);
|
|
telegram.failWith(502, 'Bad Gateway', user.chatId);
|
|
|
|
await drainFor(user.id);
|
|
|
|
const row = await deliveryOf(deliveryId);
|
|
expect(row.status).toBe('PENDING');
|
|
expect(row.attempts).toBe(1);
|
|
expect(row.availableAt.getTime()).toBeGreaterThan(Date.now());
|
|
expect(telegram.sentTo(user.chatId)).toHaveLength(0);
|
|
});
|
|
|
|
it('keeps 1.1 s between two messages to the same chat, in order', async () => {
|
|
const user = await createUser('spacing', []);
|
|
const first = await queue(user.id, { message: 'erste' });
|
|
// Another createdAt, so the order is not a tie.
|
|
await new Promise((resolve) => setTimeout(resolve, 20));
|
|
const second = await queue(user.id, { message: 'zweite' });
|
|
|
|
await drainFor(user.id);
|
|
|
|
const sent = telegram
|
|
.sentTo(user.chatId)
|
|
.map((m) => m.text.split('\n')[0]);
|
|
expect(sent).toEqual(['erste', 'zweite']);
|
|
const [a, b] = await Promise.all([
|
|
deliveryOf(first.deliveryId),
|
|
deliveryOf(second.deliveryId),
|
|
]);
|
|
expect(a.status).toBe('DELIVERED');
|
|
expect(b.status).toBe('DELIVERED');
|
|
expect(
|
|
(b.deliveredAt as Date).getTime() - (a.deliveredAt as Date).getTime(),
|
|
).toBeGreaterThanOrEqual(1_000);
|
|
});
|
|
});
|
|
describe('connect, disconnect and receive', () => {
|
|
const PASSWORD = 'SuperSecret123!';
|
|
const API = '/api/v1';
|
|
const BOT = 'apicore_e2e_bot';
|
|
const OFFSET_BASE = 9_000;
|
|
const originalUsername = process.env.TELEGRAM_BOT_USERNAME;
|
|
let updates: TelegramUpdatesService;
|
|
let nextUpdateId = OFFSET_BASE;
|
|
let savedPollState: {
|
|
updateOffset: bigint;
|
|
leaseOwner: string | null;
|
|
leaseUntil: Date | null;
|
|
};
|
|
|
|
const emailOf = (label: string) =>
|
|
`telegram-e2e-${label}-${suffix}@example.com`;
|
|
|
|
async function tokenOf(label: string): Promise<string> {
|
|
return (await login(app, emailOf(label), PASSWORD)).accessToken;
|
|
}
|
|
|
|
function requestLinkToken(
|
|
accessToken: string,
|
|
body: object = { password: PASSWORD },
|
|
) {
|
|
return request(app.getHttpServer())
|
|
.post(`${API}/profile/telegram/link-token`)
|
|
.set('Authorization', `Bearer ${accessToken}`)
|
|
.send(body);
|
|
}
|
|
|
|
/**
|
|
* A link token for the user, made by the service (the route's budget of
|
|
* 10 per minute is kept for the tests of the route itself).
|
|
*/
|
|
async function newLinkToken(userId: string): Promise<string> {
|
|
const { deepLink } = await app
|
|
.get(TelegramLinkService)
|
|
.createLinkToken(userId);
|
|
return deepLink.slice(`https://t.me/${BOT}?start=`.length);
|
|
}
|
|
|
|
function say(
|
|
chatId: bigint,
|
|
text: string,
|
|
options: { chatType?: string; firstName?: string } = {},
|
|
): void {
|
|
telegram.queueUpdate({
|
|
updateId: BigInt(nextUpdateId++),
|
|
chatId,
|
|
chatType: options.chatType ?? 'private',
|
|
text,
|
|
firstName: options.firstName ?? 'Anna',
|
|
});
|
|
}
|
|
|
|
/** Fresh poll state: the offset where the next update id starts, no lease. */
|
|
async function resetPollState(): Promise<void> {
|
|
await prisma.telegramPollState.update({
|
|
where: { id: 'singleton' },
|
|
data: {
|
|
updateOffset: BigInt(nextUpdateId),
|
|
leaseOwner: null,
|
|
leaseUntil: null,
|
|
},
|
|
});
|
|
}
|
|
|
|
const repliesTo = async (chatId: bigint) =>
|
|
(
|
|
await prisma.notificationDelivery.findMany({
|
|
where: { chatId },
|
|
orderBy: { createdAt: 'asc' },
|
|
})
|
|
).map((row) => row.templateCode);
|
|
|
|
const poll = () => updates.pollOnce(NEVER_ABORTED);
|
|
|
|
/** A user without Telegram link and with Telegram switched off. */
|
|
async function createUnlinkedUser(label: string) {
|
|
const user = await createUser(label, [], {}, false);
|
|
await prisma.notificationPreference.update({
|
|
where: { userId: user.id },
|
|
data: { telegramEnabled: false },
|
|
});
|
|
return user;
|
|
}
|
|
|
|
beforeAll(async () => {
|
|
process.env.TELEGRAM_BOT_USERNAME = BOT;
|
|
updates = app.get(TelegramUpdatesService);
|
|
savedPollState = await prisma.telegramPollState.findUniqueOrThrow({
|
|
where: { id: 'singleton' },
|
|
select: { updateOffset: true, leaseOwner: true, leaseUntil: true },
|
|
});
|
|
});
|
|
|
|
afterAll(async () => {
|
|
if (originalUsername === undefined) {
|
|
delete process.env.TELEGRAM_BOT_USERNAME;
|
|
} else {
|
|
process.env.TELEGRAM_BOT_USERNAME = originalUsername;
|
|
}
|
|
await prisma.telegramPollState.update({
|
|
where: { id: 'singleton' },
|
|
data: { ...savedPollState, lastError: null },
|
|
});
|
|
await prisma.notificationDelivery.deleteMany({
|
|
where: { chatId: { gte: chatBase, lte: chatBase + 10_000n } },
|
|
});
|
|
// Logins of this block leave refresh tokens behind.
|
|
await prisma.refreshToken.deleteMany({
|
|
where: { userId: { in: userIds } },
|
|
});
|
|
});
|
|
|
|
beforeEach(async () => {
|
|
await resetPollState();
|
|
telegram.sent.length = 0;
|
|
});
|
|
|
|
it('link token -> /start links the chat once; a second /start with the same token answers LINK_INVALID', async () => {
|
|
const user = await createUnlinkedUser('connect');
|
|
const accessToken = await tokenOf('connect');
|
|
|
|
const response = await requestLinkToken(accessToken).expect(201);
|
|
// The answer holds a secret: never cached.
|
|
expect(response.headers['cache-control']).toBe('no-store');
|
|
const { deepLink, expiresAt } = response.body as {
|
|
deepLink: string;
|
|
expiresAt: string;
|
|
};
|
|
expect(deepLink).toMatch(
|
|
new RegExp(`^https://t\\.me/${BOT}\\?start=[A-Za-z0-9_-]{43}$`),
|
|
);
|
|
const token = deepLink.slice(`https://t.me/${BOT}?start=`.length);
|
|
// Ten minutes from the database clock; only the hash is stored.
|
|
const ttlMs = new Date(expiresAt).getTime() - Date.now();
|
|
expect(ttlMs).toBeGreaterThan(9 * 60_000);
|
|
expect(ttlMs).toBeLessThanOrEqual(10 * 60_000 + CLOCK_TOLERANCE_MS);
|
|
const stored = await prisma.telegramLinkToken.findMany({
|
|
where: { userId: user.id },
|
|
});
|
|
expect(stored).toHaveLength(1);
|
|
expect(stored[0].tokenHash).toBe(
|
|
createHash('sha256').update(token).digest('hex'),
|
|
);
|
|
expect(stored[0].tokenHash).not.toBe(token);
|
|
|
|
say(user.chatId, `/start ${token}`, { firstName: 'Anna' });
|
|
await expect(poll()).resolves.toBe('polled');
|
|
|
|
const link = await prisma.telegramLink.findUniqueOrThrow({
|
|
where: { userId: user.id },
|
|
});
|
|
expect(link.chatId).toBe(user.chatId);
|
|
expect(link.telegramFirstName).toBe('Anna');
|
|
const preference = await prisma.notificationPreference.findUniqueOrThrow({
|
|
where: { userId: user.id },
|
|
});
|
|
expect(preference.telegramEnabled).toBe(true);
|
|
const notice = await prisma.notification.findFirstOrThrow({
|
|
where: { userId: user.id, category: 'SYSTEM' },
|
|
});
|
|
expect(notice.message).toMatch(
|
|
/^Telegram verbunden mit „Anna“ am \d{2}\.\d{2}\.\d{4}\. Warst du das nicht\? In den Einstellungen trennen\.$/,
|
|
);
|
|
expect(notice.link).toBe('/profile');
|
|
expect(await repliesTo(user.chatId)).toEqual(['LINKED']);
|
|
const state = await prisma.telegramPollState.findUniqueOrThrow({
|
|
where: { id: 'singleton' },
|
|
});
|
|
expect(state.updateOffset).toBe(BigInt(nextUpdateId));
|
|
expect(state.lastPollAt).not.toBeNull();
|
|
|
|
// The same token again - from the same or another chat - is spent.
|
|
const stranger = chatBase + 9_001n;
|
|
say(user.chatId, `/start ${token}`);
|
|
say(stranger, `/start ${token}`);
|
|
await poll();
|
|
expect(await repliesTo(user.chatId)).toEqual(['LINKED', 'LINK_INVALID']);
|
|
expect(await repliesTo(stranger)).toEqual(['LINK_INVALID']);
|
|
expect(
|
|
await prisma.telegramLink.count({ where: { chatId: stranger } }),
|
|
).toBe(0);
|
|
});
|
|
|
|
it('a new link token makes the older open one invalid, and a chat that moves unlinks the previous user', async () => {
|
|
const first = await createUnlinkedUser('move-a');
|
|
const second = await createUnlinkedUser('move-b');
|
|
const old = await newLinkToken(first.id);
|
|
const fresh = await newLinkToken(first.id);
|
|
const other = await newLinkToken(second.id);
|
|
|
|
say(first.chatId, `/start ${old}`);
|
|
say(first.chatId, `/start ${fresh}`);
|
|
await poll();
|
|
expect(await repliesTo(first.chatId)).toEqual(['LINK_INVALID', 'LINKED']);
|
|
|
|
// User B connects from the chat of user A.
|
|
say(first.chatId, `/start ${other}`);
|
|
await poll();
|
|
|
|
const links = await prisma.telegramLink.findMany({
|
|
where: { userId: { in: [first.id, second.id] } },
|
|
});
|
|
expect(links).toHaveLength(1);
|
|
expect(links[0].userId).toBe(second.id);
|
|
expect(links[0].chatId).toBe(first.chatId);
|
|
const previous = await prisma.notificationPreference.findUniqueOrThrow({
|
|
where: { userId: first.id },
|
|
});
|
|
expect(previous.telegramEnabled).toBe(false);
|
|
const notice = await prisma.notification.findFirstOrThrow({
|
|
where: { userId: first.id, message: { contains: 'getrennt' } },
|
|
});
|
|
expect(notice.message).toBe(
|
|
'Deine Telegram-Verbindung wurde getrennt, weil der Chat mit einem anderen Konto verbunden wurde.',
|
|
);
|
|
});
|
|
|
|
it('link-token answers 409 when the bot username is not set', async () => {
|
|
await createUnlinkedUser('nobot');
|
|
const accessToken = await tokenOf('nobot');
|
|
process.env.TELEGRAM_BOT_USERNAME = '';
|
|
try {
|
|
const response = await requestLinkToken(accessToken).expect(409);
|
|
expect((response.body as { message: string }).message).toBe(
|
|
'Telegram ist nicht eingerichtet.',
|
|
);
|
|
} finally {
|
|
process.env.TELEGRAM_BOT_USERNAME = BOT;
|
|
}
|
|
});
|
|
|
|
it('only the lease holder polls; a second worker waits', async () => {
|
|
const links = app.get(TelegramLinkService);
|
|
const workerA = new TelegramUpdatesService(
|
|
prisma,
|
|
telegram as unknown as TelegramClient,
|
|
links,
|
|
);
|
|
const workerB = new TelegramUpdatesService(
|
|
prisma,
|
|
telegram as unknown as TelegramClient,
|
|
links,
|
|
);
|
|
telegram.polledOffsets.length = 0;
|
|
|
|
await expect(workerA.pollOnce(NEVER_ABORTED)).resolves.toBe('polled');
|
|
await expect(workerB.pollOnce(NEVER_ABORTED)).resolves.toBe('no-lease');
|
|
expect(telegram.polledOffsets).toHaveLength(1);
|
|
// Every round extends the holder's lease.
|
|
await expect(workerA.pollOnce(NEVER_ABORTED)).resolves.toBe('polled');
|
|
await expect(workerB.pollOnce(NEVER_ABORTED)).resolves.toBe('no-lease');
|
|
|
|
// An expired lease (database clock) passes to the next one.
|
|
await prisma.$executeRaw`UPDATE "TelegramPollState" SET "leaseUntil" = now() - interval '1 second' WHERE "id" = 'singleton'`;
|
|
await expect(workerB.pollOnce(NEVER_ABORTED)).resolves.toBe('polled');
|
|
await expect(workerA.pollOnce(NEVER_ABORTED)).resolves.toBe('no-lease');
|
|
});
|
|
|
|
it('a lease lost during the cycle rolls the update back and the next holder handles it once', async () => {
|
|
const user = await createUnlinkedUser('lostlease');
|
|
const token = await newLinkToken(user.id);
|
|
say(user.chatId, `/start ${token}`);
|
|
// Another receiver takes over right after the updates were fetched.
|
|
telegram.onGetUpdates = async () => {
|
|
await prisma.telegramPollState.update({
|
|
where: { id: 'singleton' },
|
|
data: { leaseOwner: 'someone-else' },
|
|
});
|
|
};
|
|
|
|
try {
|
|
await expect(poll()).resolves.toBe('no-lease');
|
|
} finally {
|
|
telegram.onGetUpdates = null;
|
|
}
|
|
|
|
expect(
|
|
await prisma.telegramLink.count({ where: { userId: user.id } }),
|
|
).toBe(0);
|
|
expect(await repliesTo(user.chatId)).toEqual([]);
|
|
const open = await prisma.telegramLinkToken.findFirstOrThrow({
|
|
where: { userId: user.id },
|
|
});
|
|
expect(open.consumedAt).toBeNull();
|
|
const state = await prisma.telegramPollState.findUniqueOrThrow({
|
|
where: { id: 'singleton' },
|
|
});
|
|
expect(state.updateOffset).toBe(BigInt(nextUpdateId - 1));
|
|
|
|
// The other lease ran out: the update is handled, exactly once.
|
|
await prisma.$executeRaw`UPDATE "TelegramPollState" SET "leaseUntil" = now() - interval '1 second' WHERE "id" = 'singleton'`;
|
|
await expect(poll()).resolves.toBe('polled');
|
|
await expect(poll()).resolves.toBe('polled');
|
|
expect(await repliesTo(user.chatId)).toEqual(['LINKED']);
|
|
});
|
|
|
|
it('/stop in the chat unlinks, and groups are ignored', async () => {
|
|
const user = await createUser('stopchat', []);
|
|
const group = chatBase + 9_100n;
|
|
say(group, '/stop', { chatType: 'group' });
|
|
say(group, '/start abc', { chatType: 'supergroup' });
|
|
say(user.chatId, '/stop');
|
|
await poll();
|
|
|
|
expect(
|
|
await prisma.telegramLink.count({ where: { userId: user.id } }),
|
|
).toBe(0);
|
|
const preference = await prisma.notificationPreference.findUniqueOrThrow({
|
|
where: { userId: user.id },
|
|
});
|
|
expect(preference.telegramEnabled).toBe(false);
|
|
expect(await repliesTo(user.chatId)).toEqual(['STOPPED']);
|
|
expect(await repliesTo(group)).toEqual([]);
|
|
});
|
|
|
|
it('unlink sends STOPPED and switches Telegram off', async () => {
|
|
const user = await createUser('unlink', []);
|
|
const accessToken = await tokenOf('unlink');
|
|
|
|
const view = await request(app.getHttpServer())
|
|
.get(`${API}/profile/notification-settings`)
|
|
.set('Authorization', `Bearer ${accessToken}`)
|
|
.expect(200);
|
|
expect((view.body as { telegram: object }).telegram).toMatchObject({
|
|
botNameSet: true,
|
|
botUsername: BOT,
|
|
linked: true,
|
|
firstName: 'unlink',
|
|
enabled: true,
|
|
});
|
|
expect(JSON.stringify(view.body)).not.toContain(String(user.chatId));
|
|
|
|
// A message still waiting for the old chat must never arrive later.
|
|
const waiting = await queue(user.id, { message: 'wartet noch' });
|
|
|
|
await request(app.getHttpServer())
|
|
.post(`${API}/profile/telegram/unlink`)
|
|
.set('Authorization', `Bearer ${accessToken}`)
|
|
.expect(204);
|
|
|
|
expect(
|
|
await prisma.telegramLink.count({ where: { userId: user.id } }),
|
|
).toBe(0);
|
|
const dropped = await prisma.notificationDelivery.findUniqueOrThrow({
|
|
where: { id: waiting.deliveryId },
|
|
});
|
|
expect(dropped.status).toBe('SKIPPED');
|
|
expect(dropped.lastError).toBe('unlinked');
|
|
await drainFor(user.id);
|
|
expect(telegram.sentTo(user.chatId)).toHaveLength(0);
|
|
const preference = await prisma.notificationPreference.findUniqueOrThrow({
|
|
where: { userId: user.id },
|
|
});
|
|
expect(preference.telegramEnabled).toBe(false);
|
|
expect(await repliesTo(user.chatId)).toEqual(['STOPPED']);
|
|
|
|
// Unlinking again changes nothing.
|
|
await request(app.getHttpServer())
|
|
.post(`${API}/profile/telegram/unlink`)
|
|
.set('Authorization', `Bearer ${accessToken}`)
|
|
.expect(204);
|
|
expect(await repliesTo(user.chatId)).toEqual(['STOPPED']);
|
|
await request(app.getHttpServer())
|
|
.post(`${API}/profile/telegram/unlink`)
|
|
.expect(401);
|
|
});
|
|
it('the same token claimed at once by two different chats links exactly one', async () => {
|
|
const user = await createUnlinkedUser('race-chats');
|
|
const token = await newLinkToken(user.id);
|
|
const links = app.get(TelegramLinkService);
|
|
const chatA = chatBase + 9_200n;
|
|
const chatB = chatBase + 9_201n;
|
|
|
|
const results = await Promise.all(
|
|
[chatA, chatB].map((chat) =>
|
|
prisma.$transaction((tx) =>
|
|
links.linkFromStartIn(tx, token, chat, 'Racer'),
|
|
),
|
|
),
|
|
);
|
|
|
|
expect([...results].sort()).toEqual(['LINKED', 'LINK_INVALID']);
|
|
const stored = await prisma.telegramLink.findMany({
|
|
where: { chatId: { in: [chatA, chatB] } },
|
|
});
|
|
expect(stored).toHaveLength(1);
|
|
expect(stored[0].userId).toBe(user.id);
|
|
});
|
|
|
|
it('two /start with the same token and chat in separate transactions: exactly one LINKED', async () => {
|
|
const user = await createUnlinkedUser('race-same');
|
|
const token = await newLinkToken(user.id);
|
|
const links = app.get(TelegramLinkService);
|
|
|
|
const results = await Promise.all(
|
|
[1, 2].map(() =>
|
|
prisma.$transaction((tx) =>
|
|
links.linkFromStartIn(tx, token, user.chatId, 'Racer'),
|
|
),
|
|
),
|
|
);
|
|
|
|
expect([...results].sort()).toEqual(['LINKED', 'LINK_INVALID']);
|
|
expect(
|
|
await prisma.telegramLink.count({ where: { userId: user.id } }),
|
|
).toBe(1);
|
|
expect(
|
|
await prisma.notification.count({
|
|
where: {
|
|
userId: user.id,
|
|
message: { startsWith: 'Telegram verbunden' },
|
|
},
|
|
}),
|
|
).toBe(1);
|
|
});
|
|
|
|
it('an expired token answers LINK_INVALID and links nothing', async () => {
|
|
const user = await createUnlinkedUser('expired');
|
|
const token = await newLinkToken(user.id);
|
|
await prisma.$executeRaw`UPDATE "TelegramLinkToken" SET "expiresAt" = now() - interval '1 minute' WHERE "userId" = ${user.id}`;
|
|
|
|
say(user.chatId, `/start ${token}`);
|
|
await poll();
|
|
|
|
expect(await repliesTo(user.chatId)).toEqual(['LINK_INVALID']);
|
|
expect(
|
|
await prisma.telegramLink.count({ where: { userId: user.id } }),
|
|
).toBe(0);
|
|
const row = await prisma.telegramLinkToken.findFirstOrThrow({
|
|
where: { userId: user.id },
|
|
});
|
|
expect(row.consumedAt).toBeNull();
|
|
});
|
|
|
|
it('a poison update is skipped after the attempt limit and the queue moves on', async () => {
|
|
const user = await createUser('poison', []);
|
|
const other = chatBase + 9_300n;
|
|
const links = app.get(TelegramLinkService);
|
|
const stop = jest
|
|
.spyOn(links, 'stopIn')
|
|
.mockRejectedValue(new Error('boom'));
|
|
// A receiver with an injected limit of two attempts.
|
|
const strict = new TelegramUpdatesService(
|
|
prisma,
|
|
telegram as unknown as TelegramClient,
|
|
links,
|
|
{ attempts: 2 },
|
|
);
|
|
const first = nextUpdateId;
|
|
say(user.chatId, '/stop');
|
|
say(other, 'hallo');
|
|
|
|
try {
|
|
await expect(strict.pollOnce(NEVER_ABORTED)).resolves.toBe('error');
|
|
const failing = await prisma.telegramPollState.findUniqueOrThrow({
|
|
where: { id: 'singleton' },
|
|
});
|
|
expect(failing.updateOffset).toBe(BigInt(first));
|
|
expect(failing.lastError).toBe('update not handled: Error');
|
|
|
|
await expect(strict.pollOnce(NEVER_ABORTED)).resolves.toBe('polled');
|
|
} finally {
|
|
stop.mockRestore();
|
|
}
|
|
|
|
const state = await prisma.telegramPollState.findUniqueOrThrow({
|
|
where: { id: 'singleton' },
|
|
});
|
|
expect(state.updateOffset).toBe(BigInt(nextUpdateId));
|
|
expect(state.lastError).toBe('Telegram-Nachricht übersprungen');
|
|
// The skipped /stop did nothing; the update behind it was answered.
|
|
expect(
|
|
await prisma.telegramLink.count({ where: { userId: user.id } }),
|
|
).toBe(1);
|
|
expect(await repliesTo(other)).toEqual(['HELP']);
|
|
|
|
// A clean batch afterwards clears the notice.
|
|
await strict.pollOnce(NEVER_ABORTED);
|
|
const cleared = await prisma.telegramPollState.findUniqueOrThrow({
|
|
where: { id: 'singleton' },
|
|
});
|
|
expect(cleared.lastError).toBeNull();
|
|
});
|
|
|
|
it('releasing the lease at shutdown lets the next receiver in at once', async () => {
|
|
const links = app.get(TelegramLinkService);
|
|
const client = telegram as unknown as TelegramClient;
|
|
const first = new TelegramUpdatesService(prisma, client, links);
|
|
const second = new TelegramUpdatesService(prisma, client, links);
|
|
await expect(first.pollOnce(NEVER_ABORTED)).resolves.toBe('polled');
|
|
await expect(second.pollOnce(NEVER_ABORTED)).resolves.toBe('no-lease');
|
|
|
|
await first.releaseLease();
|
|
|
|
await expect(second.pollOnce(NEVER_ABORTED)).resolves.toBe('polled');
|
|
});
|
|
|
|
it('an old backlog never reaches a chat linked later', async () => {
|
|
const user = await createUnlinkedUser('backlog');
|
|
const backlogText = `Alter Rückstand ${suffix}`;
|
|
const waiting = await queue(user.id, { message: backlogText });
|
|
await prisma.telegramLink.create({
|
|
data: { userId: user.id, chatId: user.chatId, telegramFirstName: 'x' },
|
|
});
|
|
// Quiet hours all day keep the "Telegram verbunden" notice (SYSTEM)
|
|
// back whatever the time of day; the check below does not depend on it.
|
|
await prisma.notificationPreference.update({
|
|
where: { userId: user.id },
|
|
data: {
|
|
telegramEnabled: true,
|
|
quietFrom: '00:00',
|
|
quietUntil: '23:59',
|
|
},
|
|
});
|
|
const token = await newLinkToken(user.id);
|
|
const newChat = chatBase + 9_400n;
|
|
|
|
say(newChat, `/start ${token}`);
|
|
await poll();
|
|
|
|
expect((await deliveryOf(waiting.deliveryId)).status).toBe('SKIPPED');
|
|
await drainFor(user.id);
|
|
// The backlog text reaches neither chat; the link notice may.
|
|
for (const chatId of [newChat, user.chatId]) {
|
|
expect(
|
|
telegram
|
|
.sentTo(chatId)
|
|
.filter((sent) => sent.text.includes(backlogText)),
|
|
).toEqual([]);
|
|
}
|
|
});
|
|
|
|
// Last on purpose: it uses up the 10-per-minute budget of the route.
|
|
it('link-token requires step-up and is throttled', async () => {
|
|
await createUnlinkedUser('stepup');
|
|
const accessToken = await tokenOf('stepup');
|
|
|
|
await request(app.getHttpServer())
|
|
.post(`${API}/profile/telegram/link-token`)
|
|
.send({ password: PASSWORD })
|
|
.expect(401);
|
|
// A missing password is a malformed request.
|
|
await requestLinkToken(accessToken, {}).expect(400);
|
|
await requestLinkToken(accessToken, {
|
|
password: 'wrong-password',
|
|
}).expect(401);
|
|
expect(
|
|
await prisma.telegramLinkToken.count({
|
|
where: { user: { email: emailOf('stepup') } },
|
|
}),
|
|
).toBe(0);
|
|
|
|
// 10 per minute: the 429 arrives within 11 attempts, after 401s only.
|
|
const statuses: number[] = [];
|
|
for (
|
|
let attempt = 0;
|
|
attempt < 11 && !statuses.includes(429);
|
|
attempt++
|
|
) {
|
|
const { status } = await requestLinkToken(accessToken, {
|
|
password: 'wrong-password',
|
|
});
|
|
statuses.push(status);
|
|
}
|
|
expect(statuses.at(-1)).toBe(429);
|
|
expect(statuses.slice(0, -1).every((status) => status === 401)).toBe(
|
|
true,
|
|
);
|
|
});
|
|
});
|
|
});
|