feat: T4 wave 10 – bottling quality, phenology and honey chain UI, Telegram link/receive and ops
- Bottling quality: water content above the legal limit (strict >) answers 409 CONFIRMATION_REQUIRED bound to the measured value; prefill for jar size, best-before (shelfLifeMonths, month-end clamping) and water content; impossible dates give 400; water-content checks in src/common/quality. - Frontend phenology: "Entwicklung melden" sheet, IndexedDB v4 queue, season header on the apiary page, confirmed revoke for all apiaries. - Frontend honey chain: "Vorschlag aus …" hints with explicit take-over, prefill status note, movement ledger with withdraw/correct dialogs. - Telegram: link tokens (hashed, single use), private-chat receiver with lease and offset CAS, bounded poison-update skip, Retry-After, pending deliveries skipped on disconnect; system API for deliveries and retry; worker-only secret wiring and manifest policy tests. - ARCHITECTURE.md §7, §9, §10, §11.11–11.13, Annahmen 51–55. Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
This commit is contained in:
Nick SzaunigandClaude Opus 5.5 committed
1 parent
16cc557462
commit
fab7f7e12f
91 files changed
+8463
-104
No files matched your search
@@ -286,7 +286,7 @@ describe('TelegramClient', () => {
|
||||
}),
|
||||
);
|
||||
|
||||
const updates = await client.getUpdates(100n);
|
||||
const { updates, highestUpdateId } = await client.getUpdates(100n);
|
||||
|
||||
const [url, init] = fetchMock.mock.calls[0] as [string, RequestInit];
|
||||
expect(url).toBe(`https://api.telegram.org/bot${TOKEN}/getUpdates`);
|
||||
@@ -295,6 +295,8 @@ describe('TelegramClient', () => {
|
||||
timeout: 25,
|
||||
allowed_updates: ['message'],
|
||||
});
|
||||
// The dropped update 101 is not in the list, but 102 is the highest.
|
||||
expect(highestUpdateId).toBe(102n);
|
||||
expect(updates).toEqual([
|
||||
{
|
||||
updateId: 100n,
|
||||
@@ -313,9 +315,29 @@ describe('TelegramClient', () => {
|
||||
]);
|
||||
});
|
||||
|
||||
it('reports the highest update_id even when every update was dropped', async () => {
|
||||
fetchMock.mockResolvedValue(
|
||||
jsonResponse({
|
||||
ok: true,
|
||||
result: [
|
||||
{ update_id: 7, edited_message: {} },
|
||||
{ update_id: 9, callback_query: {} },
|
||||
],
|
||||
}),
|
||||
);
|
||||
|
||||
await expect(client.getUpdates(7n)).resolves.toEqual({
|
||||
updates: [],
|
||||
highestUpdateId: 9n,
|
||||
});
|
||||
});
|
||||
|
||||
it('returns an empty list when nothing arrived', async () => {
|
||||
fetchMock.mockResolvedValue(jsonResponse({ ok: true, result: [] }));
|
||||
await expect(client.getUpdates(0n)).resolves.toEqual([]);
|
||||
await expect(client.getUpdates(0n)).resolves.toEqual({
|
||||
updates: [],
|
||||
highestUpdateId: null,
|
||||
});
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
@@ -27,6 +27,17 @@ export interface TelegramUpdate {
|
||||
firstName: string | null;
|
||||
}
|
||||
|
||||
/**
|
||||
* What one getUpdates call returned: the updates this code understands and
|
||||
* the highest raw update_id of the answer, also of updates that were dropped
|
||||
* (no chat, no id of the right type). The receiver advances the offset past
|
||||
* it, so a dropped update is never fetched again.
|
||||
*/
|
||||
export interface TelegramUpdateBatch {
|
||||
updates: TelegramUpdate[];
|
||||
highestUpdateId: bigint | null;
|
||||
}
|
||||
|
||||
// eslint-disable-next-line no-control-regex
|
||||
const CONTROL_CHARACTERS = /[\u0000-\u0008\u000b-\u001f\u007f-\u009f]/g;
|
||||
// Zero-width characters, bidi marks, overrides and isolates: invisible, and
|
||||
@@ -104,6 +115,17 @@ function retryAfterOf(parsed: JsonObject): number | null {
|
||||
);
|
||||
}
|
||||
|
||||
function highestRawUpdateId(raw: unknown[]): bigint | null {
|
||||
let highest: bigint | null = null;
|
||||
for (const entry of raw) {
|
||||
const id = asInteger(asObject(entry)?.update_id);
|
||||
if (id !== null && (highest === null || id > highest)) {
|
||||
highest = id;
|
||||
}
|
||||
}
|
||||
return highest;
|
||||
}
|
||||
|
||||
function parseUpdate(raw: unknown): TelegramUpdate | null {
|
||||
const update = asObject(raw);
|
||||
const updateId = asInteger(update?.update_id);
|
||||
@@ -166,7 +188,7 @@ export class TelegramClient {
|
||||
async getUpdates(
|
||||
offset: bigint,
|
||||
signal?: AbortSignal,
|
||||
): Promise<TelegramUpdate[]> {
|
||||
): Promise<TelegramUpdateBatch> {
|
||||
const result = await this.call(
|
||||
'getUpdates',
|
||||
{
|
||||
@@ -180,9 +202,12 @@ export class TelegramClient {
|
||||
if (!Array.isArray(result)) {
|
||||
throw new TelegramApiError('getUpdates', null, INVALID_RESPONSE, null);
|
||||
}
|
||||
return result
|
||||
.map(parseUpdate)
|
||||
.filter((update): update is TelegramUpdate => update !== null);
|
||||
return {
|
||||
updates: result
|
||||
.map(parseUpdate)
|
||||
.filter((update): update is TelegramUpdate => update !== null),
|
||||
highestUpdateId: highestRawUpdateId(result),
|
||||
};
|
||||
}
|
||||
|
||||
private jsonNumber(method: string, value: bigint): number {
|
||||
|
||||
@@ -0,0 +1,435 @@
|
||||
import { createHash } from 'node:crypto';
|
||||
import { ConflictException } from '@nestjs/common';
|
||||
import type { NotificationPreferencesService } from '../notifications/notification-preferences.service';
|
||||
import type { NotificationsService } from '../notifications/notifications.service';
|
||||
import type { PrismaService } from '../prisma/prisma.service';
|
||||
import * as rowLock from '../prisma/row-lock';
|
||||
import {
|
||||
LINK_TOKEN_BYTES,
|
||||
LINK_TOKEN_TTL_MS,
|
||||
TelegramLinkService,
|
||||
} from './telegram-link.service';
|
||||
|
||||
jest.mock('../prisma/row-lock');
|
||||
|
||||
const lockRowsForUpdate = jest.mocked(rowLock.lockRowsForUpdate);
|
||||
|
||||
const CHAT = 4_711_000_001n;
|
||||
const TOKEN = 'A'.repeat(43);
|
||||
const TOKEN_HASH = createHash('sha256').update(TOKEN).digest('hex');
|
||||
|
||||
interface FakeTx {
|
||||
$queryRaw: jest.Mock;
|
||||
$executeRaw: jest.Mock;
|
||||
telegramLinkToken: { create: jest.Mock; findUnique: jest.Mock };
|
||||
telegramLink: {
|
||||
findUnique: jest.Mock;
|
||||
delete: jest.Mock;
|
||||
deleteMany: jest.Mock;
|
||||
create: jest.Mock;
|
||||
};
|
||||
notificationPreference: { updateMany: jest.Mock };
|
||||
notificationDelivery: { create: jest.Mock; updateMany: jest.Mock };
|
||||
user: { findFirst: jest.Mock };
|
||||
}
|
||||
|
||||
/** Flattens a tagged-template call to its SQL text and bound values. */
|
||||
function sqlOf(call: unknown[]): { text: string; values: unknown[] } {
|
||||
const [first, ...values] = call as [unknown, ...unknown[]];
|
||||
if (Array.isArray(first)) {
|
||||
return { text: (first as string[]).join('?'), values };
|
||||
}
|
||||
const sql = first as { strings: string[]; values: unknown[] };
|
||||
return { text: sql.strings.join('?'), values: sql.values };
|
||||
}
|
||||
|
||||
/** The arguments of one call of a mock (negative: counted from the end). */
|
||||
function callOf(mock: jest.Mock, index: number): unknown[] {
|
||||
const calls = mock.mock.calls as unknown[][];
|
||||
return calls.at(index) ?? [];
|
||||
}
|
||||
|
||||
describe('TelegramLinkService', () => {
|
||||
let tx: FakeTx;
|
||||
let prisma: { $transaction: jest.Mock };
|
||||
let preferences: { getOrCreate: jest.Mock };
|
||||
let notifications: { record: jest.Mock };
|
||||
let service: TelegramLinkService;
|
||||
const originalUsername = process.env.TELEGRAM_BOT_USERNAME;
|
||||
const originalToken = process.env.TELEGRAM_BOT_TOKEN;
|
||||
|
||||
beforeEach(() => {
|
||||
jest.resetAllMocks();
|
||||
process.env.TELEGRAM_BOT_USERNAME = 'apicore_test_bot';
|
||||
delete process.env.TELEGRAM_BOT_TOKEN;
|
||||
tx = {
|
||||
$queryRaw: jest.fn().mockResolvedValue([]),
|
||||
$executeRaw: jest.fn().mockResolvedValue(1),
|
||||
telegramLinkToken: {
|
||||
create: jest.fn().mockResolvedValue({ id: 'tok-1' }),
|
||||
findUnique: jest.fn().mockResolvedValue(null),
|
||||
},
|
||||
telegramLink: {
|
||||
findUnique: jest.fn().mockResolvedValue(null),
|
||||
delete: jest.fn().mockResolvedValue({}),
|
||||
deleteMany: jest.fn().mockResolvedValue({ count: 1 }),
|
||||
create: jest.fn().mockResolvedValue({}),
|
||||
},
|
||||
notificationPreference: {
|
||||
updateMany: jest.fn().mockResolvedValue({ count: 1 }),
|
||||
},
|
||||
notificationDelivery: {
|
||||
create: jest.fn().mockResolvedValue({}),
|
||||
updateMany: jest.fn().mockResolvedValue({ count: 0 }),
|
||||
},
|
||||
user: { findFirst: jest.fn().mockResolvedValue({ id: 'u1' }) },
|
||||
};
|
||||
prisma = {
|
||||
$transaction: jest
|
||||
.fn()
|
||||
.mockImplementation((work: (t: unknown) => unknown) =>
|
||||
Promise.resolve(work(tx)),
|
||||
),
|
||||
};
|
||||
preferences = { getOrCreate: jest.fn().mockResolvedValue({}) };
|
||||
notifications = { record: jest.fn().mockResolvedValue({}) };
|
||||
service = new TelegramLinkService(
|
||||
prisma as unknown as PrismaService,
|
||||
preferences as unknown as NotificationPreferencesService,
|
||||
notifications as unknown as NotificationsService,
|
||||
);
|
||||
});
|
||||
|
||||
afterAll(() => {
|
||||
if (originalUsername === undefined) {
|
||||
delete process.env.TELEGRAM_BOT_USERNAME;
|
||||
} else {
|
||||
process.env.TELEGRAM_BOT_USERNAME = originalUsername;
|
||||
}
|
||||
if (originalToken !== undefined) {
|
||||
process.env.TELEGRAM_BOT_TOKEN = originalToken;
|
||||
}
|
||||
});
|
||||
|
||||
describe('createLinkToken', () => {
|
||||
beforeEach(() => {
|
||||
tx.$queryRaw.mockResolvedValue([
|
||||
{ expiresAt: new Date('2026-10-14T10:10:00.000Z') },
|
||||
]);
|
||||
});
|
||||
|
||||
it('creates a 43-character base64url token, stores only its SHA-256 and invalidates older open tokens', async () => {
|
||||
const result = await service.createLinkToken('u1');
|
||||
|
||||
const prefix = 'https://t.me/apicore_test_bot?start=';
|
||||
expect(result.deepLink.startsWith(prefix)).toBe(true);
|
||||
const token = result.deepLink.slice(prefix.length);
|
||||
expect(LINK_TOKEN_BYTES).toBe(32);
|
||||
expect(token).toMatch(/^[A-Za-z0-9_-]{43}$/);
|
||||
expect(result.expiresAt).toEqual(new Date('2026-10-14T10:10:00.000Z'));
|
||||
|
||||
const stored = callOf(tx.telegramLinkToken.create, 0)[0] as {
|
||||
data: { userId: string; tokenHash: string };
|
||||
};
|
||||
expect(stored.data.tokenHash).toBe(
|
||||
createHash('sha256').update(token).digest('hex'),
|
||||
);
|
||||
expect(JSON.stringify(stored)).not.toContain(token);
|
||||
// Older open tokens end in the same transaction, before the new one.
|
||||
const invalidate = sqlOf(callOf(tx.$executeRaw, 0));
|
||||
expect(invalidate.text).toMatch(/"consumedAt" = now\(\)/);
|
||||
expect(invalidate.text).toMatch(/"consumedAt" IS NULL/);
|
||||
expect(invalidate.values).toEqual(['u1']);
|
||||
expect(tx.$executeRaw.mock.invocationCallOrder[0]).toBeLessThan(
|
||||
tx.telegramLinkToken.create.mock.invocationCallOrder[0],
|
||||
);
|
||||
});
|
||||
|
||||
it('sets the expiry with the database clock, ten minutes ahead', async () => {
|
||||
await service.createLinkToken('u1');
|
||||
|
||||
expect(LINK_TOKEN_TTL_MS).toBe(600_000);
|
||||
const expiry = sqlOf(callOf(tx.$queryRaw, 0));
|
||||
expect(expiry.text).toMatch(/now\(\) \+ make_interval/);
|
||||
expect(expiry.values.some((value) => value instanceof Date)).toBe(false);
|
||||
});
|
||||
|
||||
it('locks the preference row of the user first', async () => {
|
||||
await service.createLinkToken('u1');
|
||||
|
||||
expect(preferences.getOrCreate).toHaveBeenCalledWith('u1', tx);
|
||||
expect(lockRowsForUpdate).toHaveBeenCalledWith(
|
||||
tx,
|
||||
'NotificationPreference',
|
||||
['u1'],
|
||||
);
|
||||
expect(lockRowsForUpdate.mock.invocationCallOrder[0]).toBeLessThan(
|
||||
tx.$executeRaw.mock.invocationCallOrder[0],
|
||||
);
|
||||
});
|
||||
|
||||
it('createLinkToken answers 409 only when the bot username is empty, not when the token is unset', async () => {
|
||||
// Token unset (the API pod never has it) but username set: works.
|
||||
await expect(service.createLinkToken('u1')).resolves.toBeDefined();
|
||||
|
||||
process.env.TELEGRAM_BOT_USERNAME = '';
|
||||
await expect(service.createLinkToken('u1')).rejects.toThrow(
|
||||
new ConflictException('Telegram ist nicht eingerichtet.'),
|
||||
);
|
||||
// Even with a valid token: no username, no link.
|
||||
process.env.TELEGRAM_BOT_TOKEN =
|
||||
'123456789:AAEhBOweik6ad9r_QXMENQjcrGbqCr4K-RM';
|
||||
await expect(service.createLinkToken('u1')).rejects.toBeInstanceOf(
|
||||
ConflictException,
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
describe('linkFromStartIn', () => {
|
||||
/** A valid, unused token of user u1. */
|
||||
function validToken(userId = 'u1'): void {
|
||||
tx.telegramLinkToken.findUnique.mockResolvedValue({ userId });
|
||||
tx.$queryRaw.mockResolvedValue([{ userId }]);
|
||||
}
|
||||
|
||||
it('an expired or used token answers LINK_INVALID without naming an account', async () => {
|
||||
tx.telegramLinkToken.findUnique.mockResolvedValue({ userId: 'u1' });
|
||||
tx.$queryRaw.mockResolvedValue([]); // CAS found nothing: used or expired
|
||||
|
||||
const result = await service.linkFromStartIn(
|
||||
tx as never,
|
||||
TOKEN,
|
||||
CHAT,
|
||||
'Anna',
|
||||
);
|
||||
|
||||
expect(result).toBe('LINK_INVALID');
|
||||
expect(tx.telegramLink.create).not.toHaveBeenCalled();
|
||||
expect(notifications.record).not.toHaveBeenCalled();
|
||||
expect(tx.notificationPreference.updateMany).not.toHaveBeenCalled();
|
||||
const cas = sqlOf(callOf(tx.$queryRaw, 0));
|
||||
expect(cas.text).toMatch(/"consumedAt" IS NULL/);
|
||||
expect(cas.text).toMatch(/"expiresAt" > now\(\)/);
|
||||
expect(cas.text).toMatch(/RETURNING "userId"/);
|
||||
expect(cas.values).toEqual([TOKEN_HASH]);
|
||||
});
|
||||
|
||||
it('an unknown or malformed token answers LINK_INVALID without a lookup of its format', async () => {
|
||||
await expect(
|
||||
service.linkFromStartIn(tx as never, 'short', CHAT, null),
|
||||
).resolves.toBe('LINK_INVALID');
|
||||
tx.telegramLinkToken.findUnique.mockResolvedValue(null);
|
||||
await expect(
|
||||
service.linkFromStartIn(tx as never, TOKEN, CHAT, null),
|
||||
).resolves.toBe('LINK_INVALID');
|
||||
expect(tx.telegramLink.create).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('a new link shows an in-app notice with the Telegram first name', async () => {
|
||||
validToken();
|
||||
|
||||
const result = await service.linkFromStartIn(
|
||||
tx as never,
|
||||
TOKEN,
|
||||
CHAT,
|
||||
' Anna\u0007 \u202e Beck ',
|
||||
);
|
||||
|
||||
expect(result).toBe('LINKED');
|
||||
expect(tx.telegramLink.create).toHaveBeenCalledWith({
|
||||
data: { userId: 'u1', chatId: CHAT, telegramFirstName: 'Anna Beck' },
|
||||
});
|
||||
expect(tx.notificationPreference.updateMany).toHaveBeenCalledWith({
|
||||
where: { userId: 'u1' },
|
||||
data: { telegramEnabled: true },
|
||||
});
|
||||
const [notice, noticeTx] = notifications.record.mock.calls[0] as [
|
||||
{ userId: string; category: string; message: string; link: string },
|
||||
unknown,
|
||||
];
|
||||
expect(noticeTx).toBe(tx);
|
||||
expect(notice.userId).toBe('u1');
|
||||
expect(notice.category).toBe('SYSTEM');
|
||||
expect(notice.link).toBe('/profile');
|
||||
expect(notice.message).toMatch(
|
||||
/^Telegram verbunden mit „Anna Beck“ am \d{2}\.\d{2}\.\d{4}\. Warst du das nicht\? In den Einstellungen trennen\.$/,
|
||||
);
|
||||
});
|
||||
|
||||
it('cuts a long first name to 64 characters', async () => {
|
||||
validToken();
|
||||
|
||||
await service.linkFromStartIn(tx as never, TOKEN, CHAT, 'x'.repeat(200));
|
||||
|
||||
const data = (
|
||||
callOf(tx.telegramLink.create, 0)[0] as {
|
||||
data: { telegramFirstName: string };
|
||||
}
|
||||
).data;
|
||||
expect(data.telegramFirstName).toHaveLength(64);
|
||||
});
|
||||
|
||||
it('a chat that moves to another user unlinks the previous user with an in-app notice', async () => {
|
||||
validToken('u2');
|
||||
tx.telegramLink.findUnique.mockResolvedValue({
|
||||
userId: 'u1',
|
||||
chatId: CHAT,
|
||||
});
|
||||
|
||||
const result = await service.linkFromStartIn(
|
||||
tx as never,
|
||||
TOKEN,
|
||||
CHAT,
|
||||
'Ben',
|
||||
);
|
||||
|
||||
expect(result).toBe('LINKED');
|
||||
// Both users are locked in one statement, after their rows exist.
|
||||
expect(preferences.getOrCreate).toHaveBeenCalledWith('u1', tx);
|
||||
expect(preferences.getOrCreate).toHaveBeenCalledWith('u2', tx);
|
||||
expect(lockRowsForUpdate).toHaveBeenCalledTimes(1);
|
||||
expect(lockRowsForUpdate).toHaveBeenCalledWith(
|
||||
tx,
|
||||
'NotificationPreference',
|
||||
expect.arrayContaining(['u1', 'u2']),
|
||||
);
|
||||
expect(tx.telegramLink.delete).toHaveBeenCalledWith({
|
||||
where: { userId: 'u1' },
|
||||
});
|
||||
expect(tx.notificationPreference.updateMany).toHaveBeenCalledWith({
|
||||
where: { userId: 'u1' },
|
||||
data: { telegramEnabled: false },
|
||||
});
|
||||
const notices = notifications.record.mock.calls.map(
|
||||
([input]) => input as { userId: string; message: string },
|
||||
);
|
||||
expect(notices.find((notice) => notice.userId === 'u1')?.message).toBe(
|
||||
'Deine Telegram-Verbindung wurde getrennt, weil der Chat mit einem anderen Konto verbunden wurde.',
|
||||
);
|
||||
expect(notices.find((notice) => notice.userId === 'u2')?.message).toMatch(
|
||||
/^Telegram verbunden mit „Ben“/,
|
||||
);
|
||||
// The old link goes before the new one: the chat id is unique.
|
||||
expect(tx.telegramLink.delete.mock.invocationCallOrder[0]).toBeLessThan(
|
||||
tx.telegramLink.create.mock.invocationCallOrder[0],
|
||||
);
|
||||
});
|
||||
|
||||
it('reads the link of the chat before and after taking the locks and refuses a chat that changed owner', async () => {
|
||||
validToken('u2');
|
||||
tx.telegramLink.findUnique
|
||||
.mockResolvedValueOnce(null) // peek: free
|
||||
.mockResolvedValueOnce({ userId: 'u9', chatId: CHAT }); // after lock: taken
|
||||
|
||||
await expect(
|
||||
service.linkFromStartIn(tx as never, TOKEN, CHAT, 'Ben'),
|
||||
).rejects.toThrow('changed owner');
|
||||
expect(tx.telegramLink.create).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('an inactive user answers LINK_INVALID', async () => {
|
||||
validToken();
|
||||
tx.user.findFirst.mockResolvedValue(null);
|
||||
|
||||
await expect(
|
||||
service.linkFromStartIn(tx as never, TOKEN, CHAT, 'Anna'),
|
||||
).resolves.toBe('LINK_INVALID');
|
||||
expect(tx.telegramLink.create).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('locks first, reads the link afterwards', async () => {
|
||||
validToken();
|
||||
|
||||
await service.linkFromStartIn(tx as never, TOKEN, CHAT, 'Anna');
|
||||
|
||||
const lockOrder = lockRowsForUpdate.mock.invocationCallOrder[0];
|
||||
const casOrder = tx.$queryRaw.mock.invocationCallOrder[0];
|
||||
expect(lockOrder).toBeLessThan(casOrder);
|
||||
});
|
||||
});
|
||||
|
||||
describe('stopIn / unlink', () => {
|
||||
it('stopIn deletes the link of the chat and switches Telegram off under the preference lock', async () => {
|
||||
tx.telegramLink.findUnique.mockResolvedValue({
|
||||
userId: 'u1',
|
||||
chatId: CHAT,
|
||||
});
|
||||
|
||||
await service.stopIn(tx as never, CHAT);
|
||||
|
||||
expect(lockRowsForUpdate).toHaveBeenCalledWith(
|
||||
tx,
|
||||
'NotificationPreference',
|
||||
['u1'],
|
||||
);
|
||||
expect(tx.telegramLink.delete).toHaveBeenCalledWith({
|
||||
where: { userId: 'u1' },
|
||||
});
|
||||
expect(tx.notificationPreference.updateMany).toHaveBeenCalledWith({
|
||||
where: { userId: 'u1' },
|
||||
data: { telegramEnabled: false },
|
||||
});
|
||||
});
|
||||
|
||||
it('stopIn for an unknown chat changes nothing', async () => {
|
||||
await service.stopIn(tx as never, CHAT);
|
||||
|
||||
expect(tx.telegramLink.delete).not.toHaveBeenCalled();
|
||||
expect(tx.notificationPreference.updateMany).not.toHaveBeenCalled();
|
||||
expect(lockRowsForUpdate).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('unlink deletes the link, switches Telegram off and queues STOPPED for the chat', async () => {
|
||||
tx.telegramLink.findUnique.mockResolvedValue({
|
||||
userId: 'u1',
|
||||
chatId: CHAT,
|
||||
});
|
||||
|
||||
await service.unlink('u1');
|
||||
|
||||
expect(preferences.getOrCreate).toHaveBeenCalledWith('u1', tx);
|
||||
expect(lockRowsForUpdate).toHaveBeenCalledWith(
|
||||
tx,
|
||||
'NotificationPreference',
|
||||
['u1'],
|
||||
);
|
||||
expect(tx.telegramLink.delete).toHaveBeenCalledWith({
|
||||
where: { userId: 'u1' },
|
||||
});
|
||||
expect(tx.notificationPreference.updateMany).toHaveBeenCalledWith({
|
||||
where: { userId: 'u1' },
|
||||
data: { telegramEnabled: false },
|
||||
});
|
||||
expect(tx.notificationDelivery.create).toHaveBeenCalledWith({
|
||||
data: { channel: 'TELEGRAM', chatId: CHAT, templateCode: 'STOPPED' },
|
||||
});
|
||||
});
|
||||
|
||||
it('every disconnect marks the pending Telegram deliveries of the user SKIPPED', async () => {
|
||||
tx.telegramLink.findUnique.mockResolvedValue({
|
||||
userId: 'u1',
|
||||
chatId: CHAT,
|
||||
});
|
||||
|
||||
await service.unlink('u1');
|
||||
await service.stopIn(tx as never, CHAT);
|
||||
|
||||
expect(tx.notificationDelivery.updateMany).toHaveBeenCalledTimes(2);
|
||||
expect(tx.notificationDelivery.updateMany).toHaveBeenCalledWith({
|
||||
where: {
|
||||
userId: 'u1',
|
||||
channel: 'TELEGRAM',
|
||||
status: { in: ['PENDING', 'PROCESSING'] },
|
||||
},
|
||||
data: { status: 'SKIPPED', lastError: 'unlinked' },
|
||||
});
|
||||
});
|
||||
|
||||
it('unlink without a link is a no-op', async () => {
|
||||
await service.unlink('u1');
|
||||
|
||||
expect(tx.telegramLink.delete).not.toHaveBeenCalled();
|
||||
expect(tx.notificationDelivery.create).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,277 @@
|
||||
import { createHash, randomBytes } from 'node:crypto';
|
||||
import { ConflictException, Injectable } from '@nestjs/common';
|
||||
import { Prisma } from '../../generated/prisma';
|
||||
import { berlinToday } from '../common/dates';
|
||||
import { notifyWorker, WAKE_CHANNELS } from '../common/worker-wakeup';
|
||||
import { assertTransactionClient } from '../prisma/db-client';
|
||||
import { PrismaService } from '../prisma/prisma.service';
|
||||
import { lockRowsForUpdate } from '../prisma/row-lock';
|
||||
import { NotificationPreferencesService } from '../notifications/notification-preferences.service';
|
||||
import { NotificationsService } from '../notifications/notifications.service';
|
||||
import { telegramConfig } from './telegram-config';
|
||||
import type { TelegramTemplateCode } from './telegram-texts';
|
||||
|
||||
export const LINK_TOKEN_TTL_MS = 600_000;
|
||||
export const LINK_TOKEN_BYTES = 32;
|
||||
|
||||
/** 32 random bytes as base64url. */
|
||||
const LINK_TOKEN_PATTERN = /^[A-Za-z0-9_-]{43}$/;
|
||||
const MAX_FIRST_NAME_CHARS = 64;
|
||||
const NOT_CONFIGURED_MESSAGE = 'Telegram ist nicht eingerichtet.';
|
||||
const UNLINKED_NOTICE =
|
||||
'Deine Telegram-Verbindung wurde getrennt, weil der Chat mit einem anderen Konto verbunden wurde.';
|
||||
const PROFILE_LINK = '/profile';
|
||||
const UNKNOWN_FIRST_NAME = 'Telegram';
|
||||
|
||||
export type LinkResult = 'LINKED' | 'LINK_INVALID';
|
||||
|
||||
// eslint-disable-next-line no-control-regex
|
||||
const CONTROL_CHARACTERS = /[\u0000-\u001f\u007f-\u009f]/g;
|
||||
const INVISIBLE_CHARACTERS = /[\u200b-\u200f\u202a-\u202e\u2066-\u2069]/g;
|
||||
|
||||
function hashToken(token: string): string {
|
||||
return createHash('sha256').update(token).digest('hex');
|
||||
}
|
||||
|
||||
/**
|
||||
* The first name Telegram reports: trimmed, no control or invisible
|
||||
* characters, whitespace collapsed, at most 64 characters (by code point).
|
||||
* It ends up in a notice of the app, so nothing else may ride along.
|
||||
*/
|
||||
function sanitizeFirstName(name: string | null): string | null {
|
||||
if (name === null) {
|
||||
return null;
|
||||
}
|
||||
const clean = name
|
||||
.replace(CONTROL_CHARACTERS, ' ')
|
||||
.replace(INVISIBLE_CHARACTERS, ' ')
|
||||
.replace(/\s+/g, ' ')
|
||||
.trim();
|
||||
const characters = [...clean].slice(0, MAX_FIRST_NAME_CHARS).join('').trim();
|
||||
return characters === '' ? null : characters;
|
||||
}
|
||||
|
||||
/** 2026-10-14 -> 14.10.2026 */
|
||||
function germanDate(day: string): string {
|
||||
const [year, month, date] = day.split('-');
|
||||
return `${date}.${month}.${year}`;
|
||||
}
|
||||
|
||||
function uniqueIds(ids: readonly (string | undefined)[]): string[] {
|
||||
return [...new Set(ids.filter((id): id is string => id !== undefined))];
|
||||
}
|
||||
|
||||
/**
|
||||
* Connects and disconnects a user's private Telegram chat (Spec 9a.5).
|
||||
* The one-time link token lives 10 minutes and is stored only as SHA-256.
|
||||
* Every change of a connection takes the preference row of each user it
|
||||
* touches first (the sender's 403 path does the same), in one statement in
|
||||
* ascending key order, and reads link and preference only afterwards.
|
||||
* Never logs a token, a link code or a chat id.
|
||||
*/
|
||||
@Injectable()
|
||||
export class TelegramLinkService {
|
||||
constructor(
|
||||
private readonly prisma: PrismaService,
|
||||
private readonly preferences: NotificationPreferencesService,
|
||||
private readonly notifications: NotificationsService,
|
||||
) {}
|
||||
|
||||
/**
|
||||
* A fresh link token for the user; older open tokens end. 409 exactly
|
||||
* when no bot username is configured: the API pod never has the token,
|
||||
* so it cannot be what "set up" means here.
|
||||
*/
|
||||
async createLinkToken(
|
||||
userId: string,
|
||||
): Promise<{ deepLink: string; expiresAt: Date }> {
|
||||
const { botUsername } = telegramConfig();
|
||||
if (botUsername === null) {
|
||||
throw new ConflictException(NOT_CONFIGURED_MESSAGE);
|
||||
}
|
||||
const token = randomBytes(LINK_TOKEN_BYTES).toString('base64url');
|
||||
const expiresAt = await this.prisma.$transaction(async (tx) => {
|
||||
await this.lockPreferences(tx, [userId]);
|
||||
await tx.$executeRaw`UPDATE "TelegramLinkToken" SET "consumedAt" = now() WHERE "userId" = ${userId} AND "consumedAt" IS NULL`;
|
||||
const { id } = await tx.telegramLinkToken.create({
|
||||
data: { userId, tokenHash: hashToken(token), expiresAt: new Date() },
|
||||
select: { id: true },
|
||||
});
|
||||
// The expiry comes from the database clock, like every comparison.
|
||||
const [row] = await tx.$queryRaw<{ expiresAt: Date }[]>`
|
||||
UPDATE "TelegramLinkToken"
|
||||
SET "expiresAt" = now() + make_interval(secs => ${LINK_TOKEN_TTL_MS / 1000}::double precision)
|
||||
WHERE "id" = ${id}
|
||||
RETURNING "expiresAt"`;
|
||||
return row.expiresAt;
|
||||
});
|
||||
return {
|
||||
deepLink: `https://t.me/${botUsername}?start=${token}`,
|
||||
expiresAt,
|
||||
};
|
||||
}
|
||||
|
||||
/** Disconnects in the app: the link goes, Telegram is off, the chat hears STOPPED. */
|
||||
async unlink(userId: string): Promise<void> {
|
||||
await this.prisma.$transaction(async (tx) => {
|
||||
await this.lockPreferences(tx, [userId]);
|
||||
const link = await tx.telegramLink.findUnique({ where: { userId } });
|
||||
if (!link) {
|
||||
return;
|
||||
}
|
||||
await this.removeLinkIn(tx, link.userId);
|
||||
await this.queueReplyIn(tx, link.chatId, 'STOPPED');
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* /start <token> from a private chat (receiver transaction). Consumes the
|
||||
* token by compare-and-set, replaces the link of the user, takes the chat
|
||||
* from a previous user if needed and tells the app. Anything unusable
|
||||
* answers LINK_INVALID, never naming an account.
|
||||
*/
|
||||
async linkFromStartIn(
|
||||
tx: Prisma.TransactionClient,
|
||||
token: string,
|
||||
chatId: bigint,
|
||||
firstName: string | null,
|
||||
): Promise<LinkResult> {
|
||||
assertTransactionClient(tx, 'TelegramLinkService.linkFromStartIn');
|
||||
if (!LINK_TOKEN_PATTERN.test(token)) {
|
||||
return 'LINK_INVALID';
|
||||
}
|
||||
const tokenHash = hashToken(token);
|
||||
const candidate = await tx.telegramLinkToken.findUnique({
|
||||
where: { tokenHash },
|
||||
select: { userId: true },
|
||||
});
|
||||
if (!candidate) {
|
||||
return 'LINK_INVALID';
|
||||
}
|
||||
const peeked = await tx.telegramLink.findUnique({
|
||||
where: { chatId },
|
||||
select: { userId: true },
|
||||
});
|
||||
const userId = candidate.userId;
|
||||
await this.lockPreferences(tx, uniqueIds([userId, peeked?.userId]));
|
||||
// Everything below reads after the locks.
|
||||
const [consumed] = await tx.$queryRaw<{ userId: string }[]>`
|
||||
UPDATE "TelegramLinkToken" SET "consumedAt" = now()
|
||||
WHERE "tokenHash" = ${tokenHash} AND "consumedAt" IS NULL AND "expiresAt" > now()
|
||||
RETURNING "userId"`;
|
||||
if (!consumed) {
|
||||
return 'LINK_INVALID';
|
||||
}
|
||||
const owner = await tx.telegramLink.findUnique({ where: { chatId } });
|
||||
if (owner && owner.userId !== userId && owner.userId !== peeked?.userId) {
|
||||
// Not locked: the transaction fails and the update is handled again.
|
||||
throw new Error('Telegram chat changed owner while linking');
|
||||
}
|
||||
const active = await tx.user.findFirst({
|
||||
where: { id: userId, isActive: true, role: { status: 'ACTIVE' } },
|
||||
select: { id: true },
|
||||
});
|
||||
if (!active) {
|
||||
return 'LINK_INVALID';
|
||||
}
|
||||
if (owner && owner.userId !== userId) {
|
||||
await this.removeLinkIn(tx, owner.userId);
|
||||
await this.notifications.record(
|
||||
{
|
||||
userId: owner.userId,
|
||||
priority: 'HINWEIS',
|
||||
category: 'SYSTEM',
|
||||
message: UNLINKED_NOTICE,
|
||||
link: PROFILE_LINK,
|
||||
},
|
||||
tx,
|
||||
);
|
||||
}
|
||||
const name = sanitizeFirstName(firstName);
|
||||
// The user's own earlier chat is replaced; linkedAt starts again.
|
||||
await tx.telegramLink.deleteMany({ where: { userId } });
|
||||
await tx.telegramLink.create({
|
||||
data: { userId, chatId, telegramFirstName: name },
|
||||
});
|
||||
await tx.notificationPreference.updateMany({
|
||||
where: { userId },
|
||||
data: { telegramEnabled: true },
|
||||
});
|
||||
// The notice that exposes a link someone else used (Spec 9a.5).
|
||||
await this.notifications.record(
|
||||
{
|
||||
userId,
|
||||
priority: 'HINWEIS',
|
||||
category: 'SYSTEM',
|
||||
message: `Telegram verbunden mit „${name ?? UNKNOWN_FIRST_NAME}“ am ${germanDate(berlinToday())}. Warst du das nicht? In den Einstellungen trennen.`,
|
||||
link: PROFILE_LINK,
|
||||
},
|
||||
tx,
|
||||
);
|
||||
return 'LINKED';
|
||||
}
|
||||
|
||||
/** /stop from a chat (receiver transaction): link gone, Telegram off. */
|
||||
async stopIn(tx: Prisma.TransactionClient, chatId: bigint): Promise<void> {
|
||||
assertTransactionClient(tx, 'TelegramLinkService.stopIn');
|
||||
const peeked = await tx.telegramLink.findUnique({
|
||||
where: { chatId },
|
||||
select: { userId: true },
|
||||
});
|
||||
if (!peeked) {
|
||||
return;
|
||||
}
|
||||
await this.lockPreferences(tx, [peeked.userId]);
|
||||
const link = await tx.telegramLink.findUnique({
|
||||
where: { chatId },
|
||||
select: { userId: true },
|
||||
});
|
||||
if (link) {
|
||||
await this.removeLinkIn(tx, link.userId);
|
||||
}
|
||||
}
|
||||
|
||||
/** Queues a bot reply to a chat (it ignores quiet hours) and wakes the sender. */
|
||||
async queueReplyIn(
|
||||
tx: Prisma.TransactionClient,
|
||||
chatId: bigint,
|
||||
code: TelegramTemplateCode,
|
||||
): Promise<void> {
|
||||
assertTransactionClient(tx, 'TelegramLinkService.queueReplyIn');
|
||||
await tx.notificationDelivery.create({
|
||||
data: { channel: 'TELEGRAM', chatId, templateCode: code },
|
||||
});
|
||||
await notifyWorker(tx, WAKE_CHANNELS.notificationDeliveries);
|
||||
}
|
||||
|
||||
private async removeLinkIn(
|
||||
tx: Prisma.TransactionClient,
|
||||
userId: string,
|
||||
): Promise<void> {
|
||||
await tx.telegramLink.delete({ where: { userId } });
|
||||
await tx.notificationPreference.updateMany({
|
||||
where: { userId },
|
||||
data: { telegramEnabled: false },
|
||||
});
|
||||
// Whatever was queued for the old chat must never reach a chat linked later.
|
||||
await tx.notificationDelivery.updateMany({
|
||||
where: {
|
||||
userId,
|
||||
channel: 'TELEGRAM',
|
||||
status: { in: ['PENDING', 'PROCESSING'] },
|
||||
},
|
||||
data: { status: 'SKIPPED', lastError: 'unlinked' },
|
||||
});
|
||||
}
|
||||
|
||||
/** Rows must exist, then one statement locks them all in key order. */
|
||||
private async lockPreferences(
|
||||
tx: Prisma.TransactionClient,
|
||||
userIds: readonly string[],
|
||||
): Promise<void> {
|
||||
for (const userId of userIds) {
|
||||
await this.preferences.getOrCreate(userId, tx);
|
||||
}
|
||||
await lockRowsForUpdate(tx, 'NotificationPreference', userIds);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,557 @@
|
||||
import { Logger } from '@nestjs/common';
|
||||
import type { PrismaService } from '../prisma/prisma.service';
|
||||
import { TelegramApiError } from './telegram-api-error';
|
||||
import type {
|
||||
TelegramClient,
|
||||
TelegramUpdate,
|
||||
TelegramUpdateBatch,
|
||||
} from './telegram-client';
|
||||
import type { TelegramLinkService } from './telegram-link.service';
|
||||
import {
|
||||
HELP_REPLIES_PER_CHAT,
|
||||
MAX_UPDATE_ATTEMPTS,
|
||||
MAX_UPDATE_RETRY_MS,
|
||||
SKIPPED_UPDATE_TEXT,
|
||||
HELP_WINDOW_MS,
|
||||
LEASE_SECONDS,
|
||||
NO_LEASE_WAIT_MS,
|
||||
TelegramUpdatesService,
|
||||
} from './telegram-updates.service';
|
||||
|
||||
const TOKEN = 'B'.repeat(43);
|
||||
const PRIVATE_CHAT = 5_000_000_001n;
|
||||
const GROUP_CHAT = -1_000_000_000_001n;
|
||||
|
||||
function update(
|
||||
updateId: number,
|
||||
text: string | null,
|
||||
options: { chatId?: bigint; chatType?: string; firstName?: string } = {},
|
||||
): TelegramUpdate {
|
||||
return {
|
||||
updateId: BigInt(updateId),
|
||||
chatId: options.chatId ?? PRIVATE_CHAT,
|
||||
chatType: options.chatType ?? 'private',
|
||||
text,
|
||||
firstName: options.firstName ?? 'Anna',
|
||||
};
|
||||
}
|
||||
|
||||
/** Flattens a tagged-template call to its SQL text and bound values. */
|
||||
function sqlOf(call: unknown[]): { text: string; values: unknown[] } {
|
||||
const [first, ...values] = call as [unknown, ...unknown[]];
|
||||
if (Array.isArray(first)) {
|
||||
return { text: (first as string[]).join('?'), values };
|
||||
}
|
||||
const sql = first as { strings: string[]; values: unknown[] };
|
||||
return { text: sql.strings.join('?'), values: sql.values };
|
||||
}
|
||||
|
||||
/** The arguments of one call of a mock (negative: counted from the end). */
|
||||
function callOf(mock: jest.Mock, index: number): unknown[] {
|
||||
const calls = mock.mock.calls as unknown[][];
|
||||
return calls.at(index) ?? [];
|
||||
}
|
||||
|
||||
describe('TelegramUpdatesService', () => {
|
||||
let prisma: {
|
||||
$queryRaw: jest.Mock;
|
||||
$executeRaw: jest.Mock;
|
||||
$transaction: jest.Mock;
|
||||
};
|
||||
let tx: { $executeRaw: jest.Mock };
|
||||
let client: { isConfigured: jest.Mock; getUpdates: jest.Mock };
|
||||
let links: {
|
||||
linkFromStartIn: jest.Mock;
|
||||
stopIn: jest.Mock;
|
||||
queueReplyIn: jest.Mock;
|
||||
};
|
||||
let service: TelegramUpdatesService;
|
||||
let signal: AbortSignal;
|
||||
|
||||
/** The lease is free and the stored offset is `offset`. */
|
||||
function leaseGranted(offset: bigint | null = 7n): void {
|
||||
prisma.$queryRaw.mockResolvedValue(
|
||||
offset === null ? [] : [{ updateOffset: offset }],
|
||||
);
|
||||
}
|
||||
|
||||
/** What the next getUpdates calls return; the highest id counts too. */
|
||||
function giveUpdates(updates: TelegramUpdate[], highest?: number): void {
|
||||
const ids = updates.map((entry) => entry.updateId);
|
||||
const batch: TelegramUpdateBatch = {
|
||||
updates,
|
||||
highestUpdateId:
|
||||
highest !== undefined
|
||||
? BigInt(highest)
|
||||
: ids.length > 0
|
||||
? ids.reduce((a, b) => (a > b ? a : b))
|
||||
: null,
|
||||
};
|
||||
client.getUpdates.mockResolvedValue(batch);
|
||||
}
|
||||
|
||||
const replies = (): string[] =>
|
||||
(links.queueReplyIn.mock.calls as unknown[][]).map(
|
||||
(call) => call[2] as string,
|
||||
);
|
||||
|
||||
beforeEach(() => {
|
||||
jest.spyOn(Logger.prototype, 'error').mockImplementation(() => undefined);
|
||||
jest.spyOn(Logger.prototype, 'warn').mockImplementation(() => undefined);
|
||||
signal = new AbortController().signal;
|
||||
tx = { $executeRaw: jest.fn().mockResolvedValue(1) };
|
||||
prisma = {
|
||||
$queryRaw: jest.fn().mockResolvedValue([]),
|
||||
$executeRaw: jest.fn().mockResolvedValue(1),
|
||||
$transaction: jest
|
||||
.fn()
|
||||
.mockImplementation((work: (t: unknown) => unknown) =>
|
||||
Promise.resolve(work(tx)),
|
||||
),
|
||||
};
|
||||
client = {
|
||||
isConfigured: jest.fn().mockReturnValue(true),
|
||||
getUpdates: jest
|
||||
.fn()
|
||||
.mockResolvedValue({ updates: [], highestUpdateId: null }),
|
||||
};
|
||||
links = {
|
||||
linkFromStartIn: jest.fn().mockResolvedValue('LINKED'),
|
||||
stopIn: jest.fn().mockResolvedValue(undefined),
|
||||
queueReplyIn: jest.fn().mockResolvedValue(undefined),
|
||||
};
|
||||
service = new TelegramUpdatesService(
|
||||
prisma as unknown as PrismaService,
|
||||
client as unknown as TelegramClient,
|
||||
links as unknown as TelegramLinkService,
|
||||
);
|
||||
leaseGranted();
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
jest.restoreAllMocks();
|
||||
});
|
||||
|
||||
it('exposes the documented constants', () => {
|
||||
expect(LEASE_SECONDS).toBe(60);
|
||||
expect(NO_LEASE_WAIT_MS).toBe(30_000);
|
||||
expect(HELP_REPLIES_PER_CHAT).toBe(3);
|
||||
expect(HELP_WINDOW_MS).toBe(600_000);
|
||||
});
|
||||
|
||||
it('does nothing without a configured bot', async () => {
|
||||
client.isConfigured.mockReturnValue(false);
|
||||
|
||||
await expect(service.pollOnce(signal)).resolves.toBe('disabled');
|
||||
expect(prisma.$queryRaw).not.toHaveBeenCalled();
|
||||
expect(client.getUpdates).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('takes the lease with the database clock only', async () => {
|
||||
await service.pollOnce(signal);
|
||||
|
||||
const lease = sqlOf(callOf(prisma.$queryRaw, 0));
|
||||
expect(lease.text).toMatch(/"leaseUntil" = now\(\) \+ make_interval/);
|
||||
expect(lease.text).toMatch(/"id" = 'singleton'/);
|
||||
expect(lease.text).toMatch(
|
||||
/\("leaseUntil" IS NULL OR "leaseUntil" < now\(\) OR "leaseOwner" = /,
|
||||
);
|
||||
expect(lease.text).toMatch(/RETURNING "updateOffset"/);
|
||||
// Owner and lease seconds are the only parameters: no Date, no clock.
|
||||
expect(lease.values).toHaveLength(3);
|
||||
expect(lease.values.some((value) => value instanceof Date)).toBe(false);
|
||||
expect(lease.values).toContain(LEASE_SECONDS);
|
||||
});
|
||||
|
||||
it('waits without the lease and does not call Telegram', async () => {
|
||||
leaseGranted(null);
|
||||
|
||||
await expect(service.pollOnce(signal)).resolves.toBe('no-lease');
|
||||
expect(client.getUpdates).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('asks Telegram from the stored offset and reports polled', async () => {
|
||||
leaseGranted(42n);
|
||||
|
||||
await expect(service.pollOnce(signal)).resolves.toBe('polled');
|
||||
expect(client.getUpdates).toHaveBeenCalledWith(42n, signal);
|
||||
// lastPollAt is written with the database clock, lastError is cleared
|
||||
// only after the batch was applied.
|
||||
const stamp = sqlOf(callOf(prisma.$executeRaw, 0));
|
||||
expect(stamp.text).toMatch(/"lastPollAt" = now\(\)/);
|
||||
expect(stamp.text).not.toMatch(/lastError/);
|
||||
const cleared = sqlOf(callOf(prisma.$executeRaw, 1));
|
||||
expect(cleared.text).toMatch(/"lastError" = NULL/);
|
||||
});
|
||||
|
||||
it('handles /start, /stop and /hilfe and ignores groups', async () => {
|
||||
giveUpdates([
|
||||
update(10, `/start ${TOKEN}`, { firstName: 'Anna' }),
|
||||
update(11, '/stop'),
|
||||
update(12, '/hilfe'),
|
||||
update(13, '/help'),
|
||||
update(14, 'hallo'),
|
||||
update(15, '/start'),
|
||||
update(16, `/start ${TOKEN}`, {
|
||||
chatId: GROUP_CHAT,
|
||||
chatType: 'group',
|
||||
}),
|
||||
update(17, '/stop', { chatId: GROUP_CHAT, chatType: 'supergroup' }),
|
||||
]);
|
||||
links.linkFromStartIn.mockResolvedValueOnce('LINKED');
|
||||
|
||||
await expect(service.pollOnce(signal)).resolves.toBe('polled');
|
||||
|
||||
expect(links.linkFromStartIn).toHaveBeenCalledTimes(1);
|
||||
expect(links.linkFromStartIn).toHaveBeenCalledWith(
|
||||
tx,
|
||||
TOKEN,
|
||||
PRIVATE_CHAT,
|
||||
'Anna',
|
||||
);
|
||||
expect(links.stopIn).toHaveBeenCalledTimes(1);
|
||||
expect(links.stopIn).toHaveBeenCalledWith(tx, PRIVATE_CHAT);
|
||||
// LINKED, STOPPED, then HELP for /hilfe, /help (3rd reply of the
|
||||
// window) - the next ones stay silent; groups get nothing at all.
|
||||
expect(replies()).toEqual(['LINKED', 'STOPPED', 'HELP', 'HELP']);
|
||||
expect(
|
||||
(links.queueReplyIn.mock.calls as unknown[][]).every(
|
||||
(call) => call[1] === PRIVATE_CHAT,
|
||||
),
|
||||
).toBe(true);
|
||||
});
|
||||
|
||||
it('answers a start with an unusable token with LINK_INVALID and /start@bot with a token like /start', async () => {
|
||||
links.linkFromStartIn.mockResolvedValue('LINK_INVALID');
|
||||
giveUpdates([update(10, `/start@apicore_test_bot ${TOKEN}`)]);
|
||||
|
||||
await service.pollOnce(signal);
|
||||
|
||||
expect(links.linkFromStartIn).toHaveBeenCalledWith(
|
||||
tx,
|
||||
TOKEN,
|
||||
PRIVATE_CHAT,
|
||||
'Anna',
|
||||
);
|
||||
expect(replies()).toEqual(['LINK_INVALID']);
|
||||
});
|
||||
|
||||
it('answers at most 3 times per chat in 10 minutes', async () => {
|
||||
jest.useFakeTimers();
|
||||
try {
|
||||
giveUpdates([20, 21, 22, 23, 24].map((id) => update(id, 'hallo')));
|
||||
await service.pollOnce(signal);
|
||||
expect(replies()).toEqual(['HELP', 'HELP', 'HELP']);
|
||||
|
||||
// Another chat has its own count.
|
||||
links.queueReplyIn.mockClear();
|
||||
giveUpdates([update(25, 'hallo', { chatId: 5_000_000_002n })]);
|
||||
await service.pollOnce(signal);
|
||||
expect(replies()).toEqual(['HELP']);
|
||||
|
||||
// After the window the chat is answered again.
|
||||
links.queueReplyIn.mockClear();
|
||||
jest.advanceTimersByTime(HELP_WINDOW_MS + 1);
|
||||
giveUpdates([update(26, 'hallo')]);
|
||||
await service.pollOnce(signal);
|
||||
expect(replies()).toEqual(['HELP']);
|
||||
} finally {
|
||||
jest.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it('moves the offset in the same transaction as the effect', async () => {
|
||||
giveUpdates([update(10, '/stop'), update(11, 'x')]);
|
||||
leaseGranted(10n);
|
||||
|
||||
await service.pollOnce(signal);
|
||||
|
||||
expect(prisma.$transaction).toHaveBeenCalledTimes(2);
|
||||
// Per update: offset first, then the effect, both on the same tx.
|
||||
const offsets = (tx.$executeRaw.mock.calls as unknown[][]).map((call) =>
|
||||
sqlOf(call),
|
||||
);
|
||||
expect(offsets).toHaveLength(2);
|
||||
expect(offsets[0].values).toEqual(expect.arrayContaining([11n, 10n]));
|
||||
expect(offsets[1].values).toEqual(expect.arrayContaining([12n, 11n]));
|
||||
expect(tx.$executeRaw.mock.invocationCallOrder[0]).toBeLessThan(
|
||||
links.stopIn.mock.invocationCallOrder[0],
|
||||
);
|
||||
expect(links.stopIn).toHaveBeenCalledWith(tx, PRIVATE_CHAT);
|
||||
});
|
||||
|
||||
it('the offset advance is a compare-and-set on leaseOwner and the old offset; a lost lease rolls the effect back', async () => {
|
||||
giveUpdates([update(10, '/stop'), update(11, 'x')]);
|
||||
leaseGranted(10n);
|
||||
|
||||
await service.pollOnce(signal);
|
||||
|
||||
const cas = sqlOf(callOf(tx.$executeRaw, 0));
|
||||
expect(cas.text).toMatch(/UPDATE "TelegramPollState"/);
|
||||
expect(cas.text).toMatch(/SET "updateOffset" = /);
|
||||
expect(cas.text).toMatch(/"id" = 'singleton'/);
|
||||
expect(cas.text).toMatch(/"leaseOwner" = /);
|
||||
expect(cas.text).toMatch(/"leaseUntil" > now\(\)/);
|
||||
expect(cas.text).toMatch(/"updateOffset" = /);
|
||||
expect(cas.values.some((value) => value instanceof Date)).toBe(false);
|
||||
|
||||
// Lease lost (count 0): the transaction throws - effect and offset roll
|
||||
// back together - and the cycle ends; the rest is not touched.
|
||||
tx.$executeRaw.mockReset().mockResolvedValue(0);
|
||||
links.stopIn.mockClear();
|
||||
links.queueReplyIn.mockClear();
|
||||
prisma.$transaction.mockClear();
|
||||
|
||||
await expect(service.pollOnce(signal)).resolves.toBe('no-lease');
|
||||
expect(links.stopIn).not.toHaveBeenCalled();
|
||||
expect(links.queueReplyIn).not.toHaveBeenCalled();
|
||||
expect(prisma.$transaction).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('a failing effect keeps the offset and ends the cycle with error', async () => {
|
||||
giveUpdates([update(10, '/stop'), update(11, 'x')]);
|
||||
links.stopIn.mockRejectedValue(new Error('db down'));
|
||||
|
||||
await expect(service.pollOnce(signal)).resolves.toBe('error');
|
||||
expect(prisma.$transaction).toHaveBeenCalledTimes(1);
|
||||
// The failure is recorded for the system page, without ids.
|
||||
const failure = sqlOf(callOf(prisma.$executeRaw, -1));
|
||||
expect(failure.text).toMatch(/"lastError" = /);
|
||||
expect(JSON.stringify(failure.values)).not.toContain('5000000001');
|
||||
// Fixed text plus the error class: never the message of the error.
|
||||
expect(failure.values).toContain('update not handled: Error');
|
||||
expect(JSON.stringify(failure.values)).not.toContain('db down');
|
||||
});
|
||||
|
||||
it('answers conflict on 409 and stores the description as lastError', async () => {
|
||||
client.getUpdates.mockRejectedValue(
|
||||
new TelegramApiError(
|
||||
'getUpdates',
|
||||
409,
|
||||
'Conflict: terminated by other getUpdates request',
|
||||
null,
|
||||
),
|
||||
);
|
||||
|
||||
await expect(service.pollOnce(signal)).resolves.toBe('conflict');
|
||||
const stored = sqlOf(callOf(prisma.$executeRaw, 0));
|
||||
expect(stored.text).toMatch(/"lastError" = /);
|
||||
// Fixed text plus the error class, not the description from Telegram.
|
||||
expect(stored.values).toContain(
|
||||
'getUpdates conflict (409): TelegramApiError',
|
||||
);
|
||||
expect(JSON.stringify(stored.values)).not.toContain('terminated');
|
||||
});
|
||||
|
||||
it('answers error on other failures and never stores a token', async () => {
|
||||
client.getUpdates.mockRejectedValue(
|
||||
new TelegramApiError('getUpdates', null, 'timeout', null),
|
||||
);
|
||||
|
||||
await expect(service.pollOnce(signal)).resolves.toBe('error');
|
||||
expect(JSON.stringify(prisma.$executeRaw.mock.calls)).not.toMatch(
|
||||
/\d{5,}:[A-Za-z0-9_-]{20,}/,
|
||||
);
|
||||
});
|
||||
|
||||
it('an aborted poll (shutdown) records no error and hands the lease back', async () => {
|
||||
const controller = new AbortController();
|
||||
client.getUpdates.mockImplementation(() => {
|
||||
controller.abort();
|
||||
return Promise.reject(
|
||||
new TelegramApiError('getUpdates', null, 'aborted', null),
|
||||
);
|
||||
});
|
||||
|
||||
await expect(service.pollOnce(controller.signal)).resolves.toBe('error');
|
||||
|
||||
// Only the release: no lastError, no lastPollAt.
|
||||
expect(prisma.$executeRaw).toHaveBeenCalledTimes(1);
|
||||
const release = sqlOf(callOf(prisma.$executeRaw, 0));
|
||||
expect(release.text).toMatch(/SET "leaseUntil" = now\(\)/);
|
||||
expect(release.text).toMatch(/"leaseOwner" = /);
|
||||
expect(release.text).not.toMatch(/lastError/);
|
||||
});
|
||||
|
||||
it('does not touch the lease on shutdown when it never held it', async () => {
|
||||
leaseGranted(null);
|
||||
const controller = new AbortController();
|
||||
controller.abort();
|
||||
|
||||
await service.pollOnce(controller.signal);
|
||||
|
||||
expect(prisma.$executeRaw).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
describe('poison updates', () => {
|
||||
it('exposes the documented limits', () => {
|
||||
expect(MAX_UPDATE_ATTEMPTS).toBe(5);
|
||||
expect(MAX_UPDATE_RETRY_MS).toBe(600_000);
|
||||
expect(SKIPPED_UPDATE_TEXT).toBe('Telegram-Nachricht übersprungen');
|
||||
});
|
||||
|
||||
it('retries a failing update and skips it after 5 attempts in its own transaction', async () => {
|
||||
const warn = jest.spyOn(Logger.prototype, 'warn');
|
||||
giveUpdates([update(10, '/stop'), update(11, 'x')]);
|
||||
links.stopIn.mockRejectedValue(new Error('private text of Anna'));
|
||||
|
||||
for (let attempt = 1; attempt < MAX_UPDATE_ATTEMPTS; attempt++) {
|
||||
await expect(service.pollOnce(signal)).resolves.toBe('error');
|
||||
}
|
||||
expect(links.stopIn).toHaveBeenCalledTimes(MAX_UPDATE_ATTEMPTS - 1);
|
||||
tx.$executeRaw.mockClear();
|
||||
prisma.$executeRaw.mockClear();
|
||||
|
||||
// The fifth failure skips: offset past it, fixed lastError, the next one runs.
|
||||
await expect(service.pollOnce(signal)).resolves.toBe('polled');
|
||||
|
||||
const statements = (tx.$executeRaw.mock.calls as unknown[][]).map(
|
||||
(call) => sqlOf(call),
|
||||
);
|
||||
const skipOffset = statements.find(
|
||||
(statement) =>
|
||||
/SET "updateOffset"/.test(statement.text) &&
|
||||
statement.values.includes(11n),
|
||||
);
|
||||
expect(skipOffset).toBeDefined();
|
||||
expect(
|
||||
statements.some((statement) =>
|
||||
statement.values.includes(SKIPPED_UPDATE_TEXT),
|
||||
),
|
||||
).toBe(true);
|
||||
expect(warn).toHaveBeenCalledWith(
|
||||
'Telegram update skipped after repeated failures',
|
||||
);
|
||||
// No id and no text anywhere in the log.
|
||||
expect(JSON.stringify(warn.mock.calls)).not.toMatch(/10|Anna/);
|
||||
// The following update was handled, and the skip keeps its lastError.
|
||||
expect(replies()).toContain('HELP');
|
||||
expect(
|
||||
(prisma.$executeRaw.mock.calls as unknown[][]).some((call) =>
|
||||
sqlOf(call).text.includes('"lastError" = NULL'),
|
||||
),
|
||||
).toBe(false);
|
||||
});
|
||||
|
||||
it('skips after the time limit even with fewer attempts', async () => {
|
||||
jest.useFakeTimers();
|
||||
try {
|
||||
giveUpdates([update(10, '/stop')]);
|
||||
links.stopIn.mockRejectedValue(new Error('boom'));
|
||||
await expect(service.pollOnce(signal)).resolves.toBe('error');
|
||||
|
||||
jest.advanceTimersByTime(MAX_UPDATE_RETRY_MS + 1);
|
||||
|
||||
await expect(service.pollOnce(signal)).resolves.toBe('polled');
|
||||
expect(links.stopIn).toHaveBeenCalledTimes(2);
|
||||
} finally {
|
||||
jest.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it('a lost lease is no failure of the update', async () => {
|
||||
giveUpdates([update(10, '/stop')]);
|
||||
tx.$executeRaw.mockResolvedValue(0);
|
||||
|
||||
for (let attempt = 0; attempt < MAX_UPDATE_ATTEMPTS + 2; attempt++) {
|
||||
await expect(service.pollOnce(signal)).resolves.toBe('no-lease');
|
||||
}
|
||||
expect(links.stopIn).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
it('moves the offset past a batch of only unparseable updates', async () => {
|
||||
leaseGranted(7n);
|
||||
giveUpdates([], 12);
|
||||
|
||||
await expect(service.pollOnce(signal)).resolves.toBe('polled');
|
||||
|
||||
expect(prisma.$transaction).toHaveBeenCalledTimes(1);
|
||||
const cas = sqlOf(callOf(tx.$executeRaw, 0));
|
||||
expect(cas.text).toMatch(/SET "updateOffset"/);
|
||||
expect(cas.values).toEqual(expect.arrayContaining([13n, 7n]));
|
||||
expect(links.queueReplyIn).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('moves the offset past dropped updates behind the last handled one, once', async () => {
|
||||
leaseGranted(10n);
|
||||
giveUpdates([update(10, 'x')], 14);
|
||||
|
||||
await service.pollOnce(signal);
|
||||
|
||||
const offsets = (tx.$executeRaw.mock.calls as unknown[][]).map((call) =>
|
||||
sqlOf(call),
|
||||
);
|
||||
expect(offsets).toHaveLength(2);
|
||||
expect(offsets[0].values).toEqual(expect.arrayContaining([11n, 10n]));
|
||||
expect(offsets[1].values).toEqual(expect.arrayContaining([15n, 11n]));
|
||||
});
|
||||
|
||||
it('counts the reply quota only after the transaction committed', async () => {
|
||||
giveUpdates([update(10, 'hallo')]);
|
||||
// The commit fails after the reply was queued.
|
||||
prisma.$transaction.mockImplementationOnce(
|
||||
async (work: (t: unknown) => Promise<unknown>) => {
|
||||
await work(tx);
|
||||
throw new Error('commit failed');
|
||||
},
|
||||
);
|
||||
await service.pollOnce(signal);
|
||||
links.queueReplyIn.mockClear();
|
||||
|
||||
// Three helps are still allowed: the rolled-back one was not counted.
|
||||
for (const id of [11, 12, 13, 14]) {
|
||||
leaseGranted(BigInt(id));
|
||||
giveUpdates([update(id, 'hallo')]);
|
||||
await service.pollOnce(signal);
|
||||
}
|
||||
expect(replies()).toEqual(['HELP', 'HELP', 'HELP']);
|
||||
});
|
||||
|
||||
describe('429 from Telegram', () => {
|
||||
it('stays away until retry_after has passed', async () => {
|
||||
jest.useFakeTimers();
|
||||
try {
|
||||
client.getUpdates.mockRejectedValueOnce(
|
||||
new TelegramApiError('getUpdates', 429, 'Too Many Requests', 120),
|
||||
);
|
||||
await expect(service.pollOnce(signal)).resolves.toBe('error');
|
||||
const stored = sqlOf(callOf(prisma.$executeRaw, 0));
|
||||
expect(stored.values).toContain(
|
||||
'getUpdates rate limited (429): TelegramApiError',
|
||||
);
|
||||
|
||||
// Cycles inside the pause do not even ask for the lease.
|
||||
prisma.$queryRaw.mockClear();
|
||||
client.getUpdates.mockClear();
|
||||
jest.advanceTimersByTime(119_000);
|
||||
await expect(service.pollOnce(signal)).resolves.toBe('error');
|
||||
expect(prisma.$queryRaw).not.toHaveBeenCalled();
|
||||
expect(client.getUpdates).not.toHaveBeenCalled();
|
||||
|
||||
jest.advanceTimersByTime(1_001);
|
||||
await expect(service.pollOnce(signal)).resolves.toBe('polled');
|
||||
expect(client.getUpdates).toHaveBeenCalledTimes(1);
|
||||
} finally {
|
||||
jest.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it('caps the pause at one hour', async () => {
|
||||
jest.useFakeTimers();
|
||||
try {
|
||||
client.getUpdates.mockRejectedValueOnce(
|
||||
new TelegramApiError('getUpdates', 429, 'x', 86_400),
|
||||
);
|
||||
await service.pollOnce(signal);
|
||||
|
||||
jest.advanceTimersByTime(3_600_001);
|
||||
client.getUpdates.mockClear();
|
||||
await expect(service.pollOnce(signal)).resolves.toBe('polled');
|
||||
expect(client.getUpdates).toHaveBeenCalledTimes(1);
|
||||
} finally {
|
||||
jest.useRealTimers();
|
||||
}
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,446 @@
|
||||
import { randomUUID } from 'node:crypto';
|
||||
import { Injectable, Logger } from '@nestjs/common';
|
||||
import { Prisma } from '../../generated/prisma';
|
||||
import { loggableError } from '../common/loggable-error';
|
||||
import { PrismaService } from '../prisma/prisma.service';
|
||||
import { TelegramApiError } from './telegram-api-error';
|
||||
import {
|
||||
TelegramClient,
|
||||
TelegramUpdate,
|
||||
TelegramUpdateBatch,
|
||||
} from './telegram-client';
|
||||
import { TelegramLinkService } from './telegram-link.service';
|
||||
import type { TelegramTemplateCode } from './telegram-texts';
|
||||
|
||||
export const LEASE_SECONDS = 60;
|
||||
/** Loop interval: also how long a receiver without the lease (or after a 409) waits. */
|
||||
export const NO_LEASE_WAIT_MS = 30_000;
|
||||
export const HELP_REPLIES_PER_CHAT = 3;
|
||||
export const HELP_WINDOW_MS = 600_000;
|
||||
|
||||
/** An update that keeps failing is skipped after this many attempts or this long. */
|
||||
export const MAX_UPDATE_ATTEMPTS = 5;
|
||||
export const MAX_UPDATE_RETRY_MS = 600_000;
|
||||
const MAX_RATE_LIMIT_PAUSE_MS = 3_600_000;
|
||||
/** Fixed texts for the system page: no ids, no message text. */
|
||||
export const SKIPPED_UPDATE_TEXT = 'Telegram-Nachricht übersprungen';
|
||||
const MAX_LAST_ERROR_CHARS = 300;
|
||||
const START_COMMAND = /^\/([A-Za-z]+)(?:@[A-Za-z0-9_]+)?(?:\s+(\S+))?/;
|
||||
|
||||
export type PollOutcome =
|
||||
'disabled' | 'no-lease' | 'polled' | 'conflict' | 'error';
|
||||
|
||||
type Command =
|
||||
{ name: 'start'; token: string | null } | { name: 'stop' } | { name: 'help' };
|
||||
|
||||
/** The lease or the stored offset changed: nothing of this update may stay. */
|
||||
class LeaseLostError extends Error {
|
||||
constructor() {
|
||||
super('Telegram poll lease lost');
|
||||
this.name = 'LeaseLostError';
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* What may go to a log or lastError: Prisma errors by name, code and model
|
||||
* (loggableError), every other error by its class name only - a message can
|
||||
* quote user data.
|
||||
*/
|
||||
function describeError(error: unknown): string {
|
||||
if (
|
||||
error instanceof Prisma.PrismaClientKnownRequestError ||
|
||||
error instanceof Prisma.PrismaClientUnknownRequestError ||
|
||||
error instanceof Prisma.PrismaClientValidationError ||
|
||||
error instanceof Prisma.PrismaClientInitializationError ||
|
||||
error instanceof Prisma.PrismaClientRustPanicError
|
||||
) {
|
||||
return loggableError(error).message;
|
||||
}
|
||||
return error instanceof Error ? error.name : 'unknown error';
|
||||
}
|
||||
|
||||
function parseCommand(text: string | null): Command {
|
||||
const match = START_COMMAND.exec(text?.trim() ?? '');
|
||||
const name = match?.[1].toLowerCase();
|
||||
if (name === 'start') {
|
||||
return { name: 'start', token: match?.[2] ?? null };
|
||||
}
|
||||
return name === 'stop' ? { name: 'stop' } : { name: 'help' };
|
||||
}
|
||||
|
||||
/**
|
||||
* At most HELP_REPLIES_PER_CHAT replies per chat in HELP_WINDOW_MS, counted
|
||||
* in memory (there is exactly one receiver, the lease sees to that). A
|
||||
* stranger writing to the bot cannot make it talk without end. Monotonic
|
||||
* clock, so a clock step changes nothing.
|
||||
*/
|
||||
class ReplyLimiter {
|
||||
private readonly sentAt = new Map<string, number[]>();
|
||||
|
||||
/** Would one more reply to this chat be within the limit? Counts nothing. */
|
||||
allows(chatId: bigint): boolean {
|
||||
this.prune();
|
||||
return (
|
||||
(this.sentAt.get(String(chatId))?.length ?? 0) < HELP_REPLIES_PER_CHAT
|
||||
);
|
||||
}
|
||||
|
||||
/** Counts a reply that was queued and committed. */
|
||||
record(chatId: bigint): void {
|
||||
const key = String(chatId);
|
||||
this.sentAt.set(key, [...(this.sentAt.get(key) ?? []), performance.now()]);
|
||||
}
|
||||
|
||||
private prune(): void {
|
||||
const now = performance.now();
|
||||
for (const [chat, times] of this.sentAt) {
|
||||
const recent = times.filter((time) => now - time < HELP_WINDOW_MS);
|
||||
if (recent.length === 0) {
|
||||
this.sentAt.delete(chat);
|
||||
} else {
|
||||
this.sentAt.set(chat, recent);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Receives what people write to the bot (Spec 9a.5, 9a.7): long polling
|
||||
* getUpdates in the worker, no public endpoint. Exactly one receiver per
|
||||
* bot: the lease in TelegramPollState (database clock only) says who. Each
|
||||
* update runs in its own transaction together with the offset advance, which
|
||||
* is a compare-and-set on lease owner and old offset - a receiver that lost
|
||||
* the lease changes nothing, so every update takes effect once. Replies are
|
||||
* queued as deliveries (the sender talks to Telegram, not this class).
|
||||
* Never logs a token, a link code, a chat id or a message text.
|
||||
*/
|
||||
@Injectable()
|
||||
export class TelegramUpdatesService {
|
||||
private readonly logger = new Logger(TelegramUpdatesService.name);
|
||||
/** Names this receiver in the lease. */
|
||||
private readonly instanceId = randomUUID();
|
||||
private readonly replies = new ReplyLimiter();
|
||||
private holdsLease = false;
|
||||
/** Monotonic time until which Telegram asked us to stay away (429). */
|
||||
private rateLimitedUntil = 0;
|
||||
/** Failed attempts per update, in memory: a restart gives them a fresh start. */
|
||||
private readonly failures = new Map<
|
||||
bigint,
|
||||
{ attempts: number; firstAt: number }
|
||||
>();
|
||||
/** Test seam: after how many attempts / how long a failing update is skipped. */
|
||||
retryLimits = {
|
||||
attempts: MAX_UPDATE_ATTEMPTS,
|
||||
windowMs: MAX_UPDATE_RETRY_MS,
|
||||
};
|
||||
|
||||
constructor(
|
||||
private readonly prisma: PrismaService,
|
||||
private readonly client: TelegramClient,
|
||||
private readonly links: TelegramLinkService,
|
||||
) {}
|
||||
|
||||
/**
|
||||
* One cycle: take or extend the lease, fetch updates (long poll), handle
|
||||
* them one by one. 'polled' means the next cycle may start at once. A
|
||||
* cycle that ends because of shutdown hands the lease back.
|
||||
*/
|
||||
async pollOnce(signal: AbortSignal): Promise<PollOutcome> {
|
||||
try {
|
||||
return await this.cycle(signal);
|
||||
} finally {
|
||||
if (signal.aborted && this.holdsLease) {
|
||||
await this.releaseLease();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private async cycle(signal: AbortSignal): Promise<PollOutcome> {
|
||||
if (!this.client.isConfigured()) {
|
||||
return 'disabled';
|
||||
}
|
||||
if (signal.aborted || performance.now() < this.rateLimitedUntil) {
|
||||
return 'error';
|
||||
}
|
||||
const offset = await this.acquireLease();
|
||||
if (offset === null) {
|
||||
return 'no-lease';
|
||||
}
|
||||
let batch: TelegramUpdateBatch;
|
||||
try {
|
||||
batch = await this.client.getUpdates(offset, signal);
|
||||
} catch (error: unknown) {
|
||||
return this.onFetchError(error, signal);
|
||||
}
|
||||
await this.markPolled();
|
||||
return this.applyBatch(batch, offset, signal);
|
||||
}
|
||||
|
||||
/** The updates in order, then the offset past dropped ones; a clean batch clears lastError. */
|
||||
private async applyBatch(
|
||||
batch: TelegramUpdateBatch,
|
||||
offset: bigint,
|
||||
signal: AbortSignal,
|
||||
): Promise<PollOutcome> {
|
||||
let current = offset;
|
||||
let skipped = false;
|
||||
for (const update of batch.updates) {
|
||||
if (signal.aborted) {
|
||||
return 'polled';
|
||||
}
|
||||
try {
|
||||
const reply = await this.prisma.$transaction((tx) =>
|
||||
this.applyUpdate(tx, update, current),
|
||||
);
|
||||
// Counted only now: a rolled-back reply was never queued.
|
||||
if (reply !== null) {
|
||||
this.replies.record(reply);
|
||||
}
|
||||
this.failures.delete(update.updateId);
|
||||
} catch (error: unknown) {
|
||||
if (error instanceof LeaseLostError) {
|
||||
return 'no-lease';
|
||||
}
|
||||
if (!(await this.shouldSkip(update, error))) {
|
||||
return 'error';
|
||||
}
|
||||
try {
|
||||
await this.skipUpdate(update, current);
|
||||
} catch (skipError: unknown) {
|
||||
return this.onUpdateError(skipError);
|
||||
}
|
||||
skipped = true;
|
||||
}
|
||||
current = update.updateId + 1n;
|
||||
}
|
||||
const dropped = await this.skipDropped(batch, current, signal);
|
||||
if (dropped !== 'ok') {
|
||||
return dropped;
|
||||
}
|
||||
if (!skipped) {
|
||||
await this.clearLastError();
|
||||
}
|
||||
return 'polled';
|
||||
}
|
||||
|
||||
/** Dropped updates (no chat, wrong shape): move the offset past them, never fetch them again. */
|
||||
private async skipDropped(
|
||||
batch: TelegramUpdateBatch,
|
||||
current: bigint,
|
||||
signal: AbortSignal,
|
||||
): Promise<'ok' | PollOutcome> {
|
||||
const highest = batch.highestUpdateId;
|
||||
if (highest === null || highest < current || signal.aborted) {
|
||||
return 'ok';
|
||||
}
|
||||
try {
|
||||
await this.prisma.$transaction((tx) =>
|
||||
this.advanceOffsetIn(tx, current, highest + 1n),
|
||||
);
|
||||
} catch (error: unknown) {
|
||||
return this.onUpdateError(error);
|
||||
}
|
||||
return 'ok';
|
||||
}
|
||||
|
||||
/** Lease SQL, database clock only. The stored offset, or null without the lease. */
|
||||
private async acquireLease(): Promise<bigint | null> {
|
||||
const rows = await this.prisma.$queryRaw<{ updateOffset: bigint }[]>`
|
||||
UPDATE "TelegramPollState"
|
||||
SET "leaseOwner" = ${this.instanceId},
|
||||
"leaseUntil" = now() + make_interval(secs => ${LEASE_SECONDS})
|
||||
WHERE "id" = 'singleton'
|
||||
AND ("leaseUntil" IS NULL OR "leaseUntil" < now() OR "leaseOwner" = ${this.instanceId})
|
||||
RETURNING "updateOffset"`;
|
||||
this.holdsLease = rows.length > 0;
|
||||
return rows[0]?.updateOffset ?? null;
|
||||
}
|
||||
|
||||
/** Shutdown: the next receiver need not wait for the lease to run out. */
|
||||
private async releaseLease(): Promise<void> {
|
||||
this.holdsLease = false;
|
||||
try {
|
||||
await this.prisma.$executeRaw`
|
||||
UPDATE "TelegramPollState"
|
||||
SET "leaseUntil" = now()
|
||||
WHERE "id" = 'singleton' AND "leaseOwner" = ${this.instanceId}`;
|
||||
} catch (error: unknown) {
|
||||
this.logger.warn(`Telegram lease not released: ${describeError(error)}`);
|
||||
}
|
||||
}
|
||||
|
||||
private async markPolled(): Promise<void> {
|
||||
await this.prisma.$executeRaw`
|
||||
UPDATE "TelegramPollState"
|
||||
SET "lastPollAt" = now()
|
||||
WHERE "id" = 'singleton' AND "leaseOwner" = ${this.instanceId}`;
|
||||
}
|
||||
|
||||
private async clearLastError(): Promise<void> {
|
||||
await this.prisma.$executeRaw`
|
||||
UPDATE "TelegramPollState"
|
||||
SET "lastError" = NULL
|
||||
WHERE "id" = 'singleton' AND "leaseOwner" = ${this.instanceId}`;
|
||||
}
|
||||
|
||||
private async setLastError(text: string): Promise<void> {
|
||||
await this.prisma.$executeRaw`
|
||||
UPDATE "TelegramPollState"
|
||||
SET "lastError" = ${text.slice(0, MAX_LAST_ERROR_CHARS)}
|
||||
WHERE "id" = 'singleton' AND "leaseOwner" = ${this.instanceId}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* 409: another getUpdates or a webhook; wait. 429: stay away until
|
||||
* retry_after has passed (at most an hour). The texts are fixed plus the
|
||||
* error class: nothing from the answer, no token, no URL.
|
||||
*/
|
||||
private async onFetchError(
|
||||
error: unknown,
|
||||
signal: AbortSignal,
|
||||
): Promise<PollOutcome> {
|
||||
if (signal.aborted) {
|
||||
return 'error';
|
||||
}
|
||||
const status = error instanceof TelegramApiError ? error.statusCode : null;
|
||||
if (status === 429) {
|
||||
const seconds =
|
||||
error instanceof TelegramApiError ? error.retryAfterSec : null;
|
||||
this.rateLimitedUntil =
|
||||
performance.now() +
|
||||
Math.min(
|
||||
(seconds ?? NO_LEASE_WAIT_MS / 1_000) * 1_000,
|
||||
MAX_RATE_LIMIT_PAUSE_MS,
|
||||
);
|
||||
}
|
||||
let label = 'getUpdates failed';
|
||||
if (status === 409) {
|
||||
label = 'getUpdates conflict (409)';
|
||||
} else if (status === 429) {
|
||||
label = 'getUpdates rate limited (429)';
|
||||
}
|
||||
await this.setLastError(`${label}: ${describeError(error)}`);
|
||||
return status === 409 ? 'conflict' : 'error';
|
||||
}
|
||||
|
||||
private async onUpdateError(error: unknown): Promise<PollOutcome> {
|
||||
if (error instanceof LeaseLostError) {
|
||||
return 'no-lease';
|
||||
}
|
||||
// The offset stays: the update is tried again in the next cycle.
|
||||
const text = `update not handled: ${describeError(error)}`;
|
||||
this.logger.error(`Telegram ${text}`);
|
||||
await this.setLastError(text);
|
||||
return 'error';
|
||||
}
|
||||
|
||||
/**
|
||||
* Counts a failed attempt of an update; true once it failed too often or
|
||||
* for too long and has to be skipped (otherwise the failure is recorded).
|
||||
*/
|
||||
private async shouldSkip(
|
||||
update: TelegramUpdate,
|
||||
error: unknown,
|
||||
): Promise<boolean> {
|
||||
const now = performance.now();
|
||||
const known = this.failures.get(update.updateId);
|
||||
const state = {
|
||||
attempts: (known?.attempts ?? 0) + 1,
|
||||
firstAt: known?.firstAt ?? now,
|
||||
};
|
||||
this.failures.set(update.updateId, state);
|
||||
const exhausted =
|
||||
state.attempts >= this.retryLimits.attempts ||
|
||||
now - state.firstAt >= this.retryLimits.windowMs;
|
||||
if (!exhausted) {
|
||||
await this.onUpdateError(error);
|
||||
}
|
||||
return exhausted;
|
||||
}
|
||||
|
||||
/** Offset past a poison update in its own transaction, so the queue moves on. */
|
||||
private async skipUpdate(
|
||||
update: TelegramUpdate,
|
||||
oldOffset: bigint,
|
||||
): Promise<void> {
|
||||
await this.prisma.$transaction(async (tx) => {
|
||||
await this.advanceOffsetIn(tx, oldOffset, update.updateId + 1n);
|
||||
await tx.$executeRaw`
|
||||
UPDATE "TelegramPollState"
|
||||
SET "lastError" = ${SKIPPED_UPDATE_TEXT}
|
||||
WHERE "id" = 'singleton' AND "leaseOwner" = ${this.instanceId}`;
|
||||
});
|
||||
this.failures.delete(update.updateId);
|
||||
this.logger.warn('Telegram update skipped after repeated failures');
|
||||
}
|
||||
|
||||
/** Offset compare-and-set on lease owner and old offset; count 0: lease lost. */
|
||||
private async advanceOffsetIn(
|
||||
tx: Prisma.TransactionClient,
|
||||
oldOffset: bigint,
|
||||
newOffset: bigint,
|
||||
): Promise<void> {
|
||||
const advanced = await tx.$executeRaw`
|
||||
UPDATE "TelegramPollState"
|
||||
SET "updateOffset" = ${newOffset}
|
||||
WHERE "id" = 'singleton'
|
||||
AND "leaseOwner" = ${this.instanceId}
|
||||
AND "leaseUntil" > now()
|
||||
AND "updateOffset" = ${oldOffset}`;
|
||||
if (advanced === 0) {
|
||||
throw new LeaseLostError();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Offset first (a lost lease ends here), then the effect, all in `tx`.
|
||||
* Returns the chat whose reply quota must be counted once `tx` commits.
|
||||
*/
|
||||
private async applyUpdate(
|
||||
tx: Prisma.TransactionClient,
|
||||
update: TelegramUpdate,
|
||||
oldOffset: bigint,
|
||||
): Promise<bigint | null> {
|
||||
await this.advanceOffsetIn(tx, oldOffset, update.updateId + 1n);
|
||||
// Groups and channels: the bot is not for them.
|
||||
if (update.chatType !== 'private') {
|
||||
return null;
|
||||
}
|
||||
const command = parseCommand(update.text);
|
||||
if (command.name === 'stop') {
|
||||
await this.links.stopIn(tx, update.chatId);
|
||||
return this.reply(tx, update.chatId, 'STOPPED');
|
||||
}
|
||||
if (command.name === 'start' && command.token !== null) {
|
||||
const result = await this.links.linkFromStartIn(
|
||||
tx,
|
||||
command.token,
|
||||
update.chatId,
|
||||
update.firstName,
|
||||
);
|
||||
return this.reply(tx, update.chatId, result);
|
||||
}
|
||||
return this.reply(tx, update.chatId, 'HELP');
|
||||
}
|
||||
|
||||
/**
|
||||
* LINKED always goes out (it needs a valid one-time token, so no
|
||||
* stranger can trigger it); every other reply counts against the limit.
|
||||
* Returns the chat to count after the commit, or null.
|
||||
*/
|
||||
private async reply(
|
||||
tx: Prisma.TransactionClient,
|
||||
chatId: bigint,
|
||||
code: TelegramTemplateCode,
|
||||
): Promise<bigint | null> {
|
||||
if (code === 'LINKED') {
|
||||
await this.links.queueReplyIn(tx, chatId, code);
|
||||
return null;
|
||||
}
|
||||
if (!this.replies.allows(chatId)) {
|
||||
return null;
|
||||
}
|
||||
await this.links.queueReplyIn(tx, chatId, code);
|
||||
return chatId;
|
||||
}
|
||||
}
|
||||
@@ -2,10 +2,13 @@ import { Module } from '@nestjs/common';
|
||||
import { NotificationsModule } from '../notifications/notifications.module';
|
||||
import { telegramConfig } from './telegram-config';
|
||||
import { TELEGRAM_CONFIG, TelegramClient } from './telegram-client';
|
||||
import { TelegramLinkService } from './telegram-link.service';
|
||||
import { TelegramSender } from './telegram-sender';
|
||||
import { TelegramUpdatesService } from './telegram-updates.service';
|
||||
|
||||
/**
|
||||
* The Bot API client for the worker's Telegram loops. Reads the token from
|
||||
* The Bot API client for the worker's Telegram loops (send, receive) and the
|
||||
* connect/disconnect logic the profile routes use. Reads the token from
|
||||
* the environment once; without it the client reports isConfigured() false
|
||||
* and the loops do not run. The sender needs the permission resolver, which
|
||||
* the global JobPermissionsModule (worker) or PermissionsModule (API) provides:
|
||||
@@ -18,7 +21,14 @@ import { TelegramSender } from './telegram-sender';
|
||||
{ provide: TELEGRAM_CONFIG, useFactory: () => telegramConfig() },
|
||||
TelegramClient,
|
||||
TelegramSender,
|
||||
TelegramLinkService,
|
||||
TelegramUpdatesService,
|
||||
],
|
||||
exports: [
|
||||
TelegramClient,
|
||||
TelegramSender,
|
||||
TelegramLinkService,
|
||||
TelegramUpdatesService,
|
||||
],
|
||||
exports: [TelegramClient, TelegramSender],
|
||||
})
|
||||
export class TelegramModule {}
|
||||
Reference in new issue
Block a user