feat: T4b/T4c wave 9 - phenology reports, autopilot A4-A10, honey ledger, Telegram delivery
- Phenology reports API with corrections and revokes; recompute per apiary via PhenologyRecomputeRequested, allowBackward for corrections; season changes are announced in the morning digest (changes since the previous digest, each once) - Season autopilot A4-A10 with per-apiary sync (Apiary before Colony locks) and WinterCheckRecorded - Honey batch ledger at runtime: initial amount, BOTTLING bookings, manual movements, one gram rounding, untracked bottlings never book - Telegram delivery: planning, Berlin delivery times incl. DST, quiet hours, permission re-check and stale skip at send time, rate limits, cleanup of old deliveries - ARCHITECTURE.md: assumptions 47-50 and updated sections 11.x Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
This commit is contained in:
Nick SzaunigandClaude Opus 5.5 committed
1 parent
1971b2d9c4
commit
16cc557462
74 files changed
+9634
-340
No files matched your search
@@ -0,0 +1,634 @@
|
||||
import { Injectable, Logger } from '@nestjs/common';
|
||||
import { Prisma } from '../../generated/prisma';
|
||||
import { PermissionResolverService } from '../auth/permissions/permission-resolver.service';
|
||||
import { berlinToday } from '../common/dates';
|
||||
import { loggableError } from '../common/loggable-error';
|
||||
import {
|
||||
claimBatch,
|
||||
deferClaim,
|
||||
releaseStaleClaimRows,
|
||||
settleFailure,
|
||||
settleSkipped,
|
||||
settleSuccess,
|
||||
} from '../common/outbox/outbox-claims';
|
||||
import { DRAIN_MAX_ROUNDS } from '../domain-events/domain-event-dispatcher';
|
||||
import { NotificationsService } from '../notifications/notifications.service';
|
||||
import { PrismaService } from '../prisma/prisma.service';
|
||||
import { lockRowForUpdate } from '../prisma/row-lock';
|
||||
import { TelegramApiError } from './telegram-api-error';
|
||||
import { TelegramClient } from './telegram-client';
|
||||
import {
|
||||
renderTelegramText,
|
||||
TELEGRAM_TEMPLATES,
|
||||
TelegramTemplateCode,
|
||||
} from './telegram-texts';
|
||||
|
||||
export const TELEGRAM_SEND_POLL_MS = 5_000;
|
||||
export const TELEGRAM_SEND_BATCH = 25;
|
||||
export const TELEGRAM_MAX_PER_SECOND = 25;
|
||||
export const TELEGRAM_CHAT_SPACING_MS = 1_100;
|
||||
export const TELEGRAM_TOKEN_INVALID_PAUSE_MS = 900_000;
|
||||
|
||||
const TABLE = 'NotificationDelivery';
|
||||
const MIN_SEND_GAP_MS = Math.ceil(1_000 / TELEGRAM_MAX_PER_SECOND);
|
||||
const DEFAULT_RETRY_AFTER_SEC = 5;
|
||||
/** How long the memory of "last message to this chat" is kept. */
|
||||
const CHAT_MEMORY_MS = 10_000;
|
||||
|
||||
const CHAT_GONE_MESSAGE =
|
||||
'Telegram-Zustellung beendet. Der Bot wurde blockiert oder der Chat gelöscht.';
|
||||
const TOKEN_INVALID_MESSAGE = 'Telegram: Bot-Token ungültig.';
|
||||
|
||||
/** Reasons a delivery is skipped; they land in lastError, so no personal data. */
|
||||
const SKIP = {
|
||||
userInactive: 'user inactive',
|
||||
telegramDisabled: 'telegram disabled',
|
||||
notLinked: 'not linked',
|
||||
categoryOff: 'category off',
|
||||
permissionRevoked: 'permission revoked',
|
||||
stale: 'stale',
|
||||
taskDone: 'task already done',
|
||||
chatUnavailable: 'chat unavailable',
|
||||
unknownTemplate: 'unknown template',
|
||||
notificationGone: 'notification gone',
|
||||
} as const;
|
||||
|
||||
/** A claimed NotificationDelivery row, as RETURNING * yields it. */
|
||||
interface ClaimedDelivery {
|
||||
id: string;
|
||||
userId: string | null;
|
||||
chatId: bigint | null;
|
||||
notificationId: string | null;
|
||||
templateCode: string | null;
|
||||
attempts: number;
|
||||
claimedAt: Date;
|
||||
}
|
||||
|
||||
/** What the send-time check reads about a delivery, in one query. */
|
||||
interface DeliveryContext {
|
||||
userActive: boolean;
|
||||
telegramEnabled: boolean;
|
||||
categories: string[];
|
||||
detail: boolean;
|
||||
chatId: bigint | null;
|
||||
category: string;
|
||||
message: string;
|
||||
link: string | null;
|
||||
requiredPermissions: string[];
|
||||
stale: boolean;
|
||||
taskOpen: boolean;
|
||||
}
|
||||
|
||||
type Target = { skip: string } | { chatId: bigint; text: string };
|
||||
|
||||
/** 'spaced': put back only because the chat got a message a moment ago. */
|
||||
type Outcome = 'settled' | 'spaced';
|
||||
|
||||
interface RoundResult {
|
||||
claimed: number;
|
||||
spaced: number;
|
||||
}
|
||||
|
||||
export interface TelegramSendFilter {
|
||||
/** Only deliveries of these users (e2e tests: no foreign rows). */
|
||||
userIds?: readonly string[];
|
||||
}
|
||||
|
||||
function claimFilter(filter?: TelegramSendFilter): Prisma.Sql | undefined {
|
||||
if (filter?.userIds === undefined) {
|
||||
return undefined;
|
||||
}
|
||||
return filter.userIds.length === 0
|
||||
? Prisma.sql`FALSE`
|
||||
: Prisma.sql`"userId" IN (${Prisma.join([...filter.userIds])})`;
|
||||
}
|
||||
|
||||
function targetKey(delivery: ClaimedDelivery): string {
|
||||
return delivery.userId !== null
|
||||
? `user:${delivery.userId}`
|
||||
: `chat:${String(delivery.chatId)}`;
|
||||
}
|
||||
|
||||
/** The oldest delivery of each user or chat; the claim is ordered by age. */
|
||||
function splitOldestPerTarget(claimed: readonly ClaimedDelivery[]): {
|
||||
oldest: ClaimedDelivery[];
|
||||
rest: ClaimedDelivery[];
|
||||
} {
|
||||
const seen = new Set<string>();
|
||||
const oldest: ClaimedDelivery[] = [];
|
||||
const rest: ClaimedDelivery[] = [];
|
||||
for (const delivery of claimed) {
|
||||
const key = targetKey(delivery);
|
||||
if (seen.has(key)) {
|
||||
rest.push(delivery);
|
||||
} else {
|
||||
seen.add(key);
|
||||
oldest.push(delivery);
|
||||
}
|
||||
}
|
||||
return { oldest, rest };
|
||||
}
|
||||
|
||||
function sleep(ms: number, signal: AbortSignal): Promise<void> {
|
||||
return new Promise((resolve) => {
|
||||
if (signal.aborted) {
|
||||
resolve();
|
||||
return;
|
||||
}
|
||||
const done = (): void => {
|
||||
clearTimeout(timer);
|
||||
signal.removeEventListener('abort', done);
|
||||
resolve();
|
||||
};
|
||||
const timer = setTimeout(done, ms);
|
||||
signal.addEventListener('abort', done, { once: true });
|
||||
});
|
||||
}
|
||||
|
||||
function isChatGone(error: TelegramApiError): boolean {
|
||||
return (
|
||||
error.statusCode === 403 ||
|
||||
(error.statusCode === 400 && /chat not found/i.test(error.description))
|
||||
);
|
||||
}
|
||||
|
||||
const DEVELOPMENT_ORIGIN = 'http://localhost:3000';
|
||||
/** A batch stops after this long; the claim is stale after 5 minutes. */
|
||||
export const TELEGRAM_BATCH_BUDGET_MS = 120_000;
|
||||
|
||||
/**
|
||||
* Sends the queued Telegram deliveries (Spec 9a.6): claims up to 25 due
|
||||
* rows, takes the oldest of each user or chat, checks again at send time
|
||||
* that the message may still go out, and settles each row. Bot replies
|
||||
* (templateCode) skip the checks: they answer a person in a chat.
|
||||
* Run by the worker's telegram-send loop. ASSUMPTION: one worker replica
|
||||
* (the worker deployment has replicas: 1). The per-chat spacing, the global
|
||||
* rate limit and the pause after a 401 or 429 live in this pod's memory; with
|
||||
* several pods each would keep its own (the one-delivery-per-target rule in
|
||||
* SQL and Telegram's 429 answers would still protect the bot). Never logs the token, a chat id
|
||||
* or a message text.
|
||||
*/
|
||||
@Injectable()
|
||||
export class TelegramSender {
|
||||
private readonly logger = new Logger(TelegramSender.name);
|
||||
private pausedUntil = 0;
|
||||
private warnedMissingOrigin = false;
|
||||
private lastSendAt = 0;
|
||||
private readonly lastSendToChat = new Map<string, number>();
|
||||
|
||||
constructor(
|
||||
private readonly prisma: PrismaService,
|
||||
private readonly client: TelegramClient,
|
||||
private readonly permissions: PermissionResolverService,
|
||||
private readonly notifications: NotificationsService,
|
||||
) {}
|
||||
|
||||
/** One batch. Returns how many deliveries it claimed. */
|
||||
async runOnce(
|
||||
signal: AbortSignal,
|
||||
filter?: TelegramSendFilter,
|
||||
): Promise<number> {
|
||||
return (await this.round(signal, filter)).claimed;
|
||||
}
|
||||
|
||||
/**
|
||||
* Runs batches until one claims nothing, at most `maxRounds`. After a
|
||||
* batch that put messages back for the chat spacing it waits that
|
||||
* spacing, so a burst for one person does not wait for the next poll.
|
||||
*/
|
||||
async drain(
|
||||
signal: AbortSignal,
|
||||
maxRounds = DRAIN_MAX_ROUNDS,
|
||||
filter?: TelegramSendFilter,
|
||||
): Promise<number> {
|
||||
let total = 0;
|
||||
for (let round = 0; round < maxRounds && !signal.aborted; round++) {
|
||||
const result = await this.round(signal, filter);
|
||||
if (result.claimed === 0) {
|
||||
break;
|
||||
}
|
||||
total += result.claimed;
|
||||
if (result.spaced > 0) {
|
||||
await sleep(TELEGRAM_CHAT_SPACING_MS, signal);
|
||||
}
|
||||
}
|
||||
return total;
|
||||
}
|
||||
|
||||
/** Puts deliveries whose worker died back in line (or to FAILED). */
|
||||
async releaseStaleClaims(): Promise<number> {
|
||||
const released = await releaseStaleClaimRows(this.prisma, TABLE);
|
||||
const failed = released
|
||||
.filter((claim) => claim.status === 'FAILED')
|
||||
.map((claim) => claim.id);
|
||||
if (failed.length > 0) {
|
||||
this.logger.error(
|
||||
`Telegram deliveries FAILED, an expired claim used up their attempts: ${failed.join(', ')}`,
|
||||
);
|
||||
}
|
||||
return released.length;
|
||||
}
|
||||
|
||||
private async round(
|
||||
signal: AbortSignal,
|
||||
filter?: TelegramSendFilter,
|
||||
): Promise<RoundResult> {
|
||||
if (signal.aborted || this.remainingPauseMs() > 0) {
|
||||
return { claimed: 0, spaced: 0 };
|
||||
}
|
||||
const claimed = await claimBatch<ClaimedDelivery>(this.prisma, TABLE, {
|
||||
limit: TELEGRAM_SEND_BATCH,
|
||||
orderBy: 'createdAt',
|
||||
filter: claimFilter(filter),
|
||||
});
|
||||
const { oldest, rest } = splitOldestPerTarget(claimed);
|
||||
let spaced = rest.length;
|
||||
for (const delivery of rest) {
|
||||
await deferClaim(
|
||||
this.prisma,
|
||||
TABLE,
|
||||
delivery.id,
|
||||
delivery.claimedAt,
|
||||
TELEGRAM_CHAT_SPACING_MS,
|
||||
);
|
||||
}
|
||||
const started = performance.now();
|
||||
for (const delivery of oldest) {
|
||||
if (performance.now() - started > TELEGRAM_BATCH_BUDGET_MS) {
|
||||
// Out of time: the claim would go stale. Back without an attempt.
|
||||
await deferClaim(
|
||||
this.prisma,
|
||||
TABLE,
|
||||
delivery.id,
|
||||
delivery.claimedAt,
|
||||
0,
|
||||
);
|
||||
continue;
|
||||
}
|
||||
const outcome = await this.handle(delivery, signal);
|
||||
if (outcome === 'spaced') {
|
||||
spaced++;
|
||||
}
|
||||
}
|
||||
return { claimed: claimed.length, spaced };
|
||||
}
|
||||
|
||||
/** Shutdown and a paused bot put the claim back without an attempt. */
|
||||
private async handle(
|
||||
delivery: ClaimedDelivery,
|
||||
signal: AbortSignal,
|
||||
): Promise<Outcome> {
|
||||
const pauseMs = this.remainingPauseMs();
|
||||
if (signal.aborted || pauseMs > 0) {
|
||||
await deferClaim(
|
||||
this.prisma,
|
||||
TABLE,
|
||||
delivery.id,
|
||||
delivery.claimedAt,
|
||||
pauseMs,
|
||||
);
|
||||
return 'settled';
|
||||
}
|
||||
try {
|
||||
return await this.process(delivery, signal);
|
||||
} catch (error: unknown) {
|
||||
await this.failSafely(delivery, error);
|
||||
return 'settled';
|
||||
}
|
||||
}
|
||||
|
||||
private async process(
|
||||
delivery: ClaimedDelivery,
|
||||
signal: AbortSignal,
|
||||
): Promise<Outcome> {
|
||||
const target =
|
||||
delivery.templateCode !== null
|
||||
? this.templateTarget(delivery)
|
||||
: await this.notificationTarget(delivery);
|
||||
if ('skip' in target) {
|
||||
await settleSkipped(
|
||||
this.prisma,
|
||||
TABLE,
|
||||
delivery.id,
|
||||
delivery.claimedAt,
|
||||
target.skip,
|
||||
);
|
||||
return 'settled';
|
||||
}
|
||||
const waitMs = this.chatWaitMs(target.chatId);
|
||||
if (waitMs > 0) {
|
||||
await deferClaim(
|
||||
this.prisma,
|
||||
TABLE,
|
||||
delivery.id,
|
||||
delivery.claimedAt,
|
||||
waitMs,
|
||||
);
|
||||
return 'spaced';
|
||||
}
|
||||
await this.keepRateLimit(signal);
|
||||
await this.send(delivery, target, signal);
|
||||
return 'settled';
|
||||
}
|
||||
|
||||
private templateTarget(delivery: ClaimedDelivery): Target {
|
||||
const code = delivery.templateCode;
|
||||
if (
|
||||
code === null ||
|
||||
delivery.chatId === null ||
|
||||
!Object.hasOwn(TELEGRAM_TEMPLATES, code)
|
||||
) {
|
||||
return { skip: SKIP.unknownTemplate };
|
||||
}
|
||||
return {
|
||||
chatId: delivery.chatId,
|
||||
text: TELEGRAM_TEMPLATES[code as TelegramTemplateCode],
|
||||
};
|
||||
}
|
||||
|
||||
/** Spec 9a.6: everything that may have changed since the delivery was queued. */
|
||||
private async notificationTarget(delivery: ClaimedDelivery): Promise<Target> {
|
||||
const context = await this.loadContext(delivery);
|
||||
if (context === null) {
|
||||
return { skip: SKIP.notificationGone };
|
||||
}
|
||||
if (!context.userActive) {
|
||||
return { skip: SKIP.userInactive };
|
||||
}
|
||||
if (!context.telegramEnabled) {
|
||||
return { skip: SKIP.telegramDisabled };
|
||||
}
|
||||
if (context.chatId === null) {
|
||||
return { skip: SKIP.notLinked };
|
||||
}
|
||||
if (!context.categories.includes(context.category)) {
|
||||
return { skip: SKIP.categoryOff };
|
||||
}
|
||||
// Fresh, not the 30 s cache: this process never hears the invalidation
|
||||
// of the API pods, so a revoked right would keep sending for 30 s.
|
||||
const held = await this.permissions.getFreshPermissions(
|
||||
delivery.userId as string,
|
||||
);
|
||||
if (!context.requiredPermissions.every((key) => held.has(key))) {
|
||||
return { skip: SKIP.permissionRevoked };
|
||||
}
|
||||
if (context.stale) {
|
||||
return { skip: SKIP.stale };
|
||||
}
|
||||
if (!context.taskOpen) {
|
||||
return { skip: SKIP.taskDone };
|
||||
}
|
||||
return {
|
||||
chatId: context.chatId,
|
||||
text: renderTelegramText(context, context.detail, this.frontendOrigin()),
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* One query for user, settings, link, notification, age and task. Stale:
|
||||
* TASK_DUE and DIGEST only on the Berlin day they were written, every
|
||||
* other category for 24 hours. "createdAt" is a UTC timestamp without
|
||||
* zone; AT TIME ZONE 'UTC' makes it an instant first.
|
||||
*/
|
||||
private async loadContext(
|
||||
delivery: ClaimedDelivery,
|
||||
): Promise<DeliveryContext | null> {
|
||||
const rows = await this.prisma.$queryRaw<DeliveryContext[]>(Prisma.sql`
|
||||
SELECT
|
||||
(u."isActive" AND r."status" = 'ACTIVE') AS "userActive",
|
||||
COALESCE(p."telegramEnabled", FALSE) AS "telegramEnabled",
|
||||
COALESCE(p."telegramCategories"::text[], ARRAY[]::text[]) AS "categories",
|
||||
COALESCE(p."telegramDetail", TRUE) AS "detail",
|
||||
l."chatId" AS "chatId",
|
||||
n."category"::text AS "category",
|
||||
n."message" AS "message",
|
||||
n."link" AS "link",
|
||||
n."requiredPermissions" AS "requiredPermissions",
|
||||
CASE
|
||||
WHEN n."category" IN ('TASK_DUE', 'DIGEST') THEN
|
||||
to_char((n."createdAt" AT TIME ZONE 'UTC') AT TIME ZONE 'Europe/Berlin', 'YYYY-MM-DD')
|
||||
<> to_char(now() AT TIME ZONE 'Europe/Berlin', 'YYYY-MM-DD')
|
||||
ELSE now() - (n."createdAt" AT TIME ZONE 'UTC') > interval '24 hours'
|
||||
END AS "stale",
|
||||
CASE
|
||||
WHEN n."category" = 'TASK_DUE' AND n."relatedType" = 'Task' AND n."relatedId" IS NOT NULL THEN
|
||||
COALESCE((SELECT t."status" IN ('OPEN', 'IN_PROGRESS') FROM "Task" t WHERE t."id" = n."relatedId"), FALSE)
|
||||
ELSE TRUE
|
||||
END AS "taskOpen"
|
||||
FROM "NotificationDelivery" d
|
||||
JOIN "Notification" n ON n."id" = d."notificationId"
|
||||
JOIN "User" u ON u."id" = d."userId"
|
||||
JOIN "Role" r ON r."id" = u."roleId"
|
||||
LEFT JOIN "NotificationPreference" p ON p."userId" = u."id"
|
||||
LEFT JOIN "TelegramLink" l ON l."userId" = u."id"
|
||||
WHERE d."id" = ${delivery.id}`);
|
||||
return rows[0] ?? null;
|
||||
}
|
||||
|
||||
private async send(
|
||||
delivery: ClaimedDelivery,
|
||||
target: { chatId: bigint; text: string },
|
||||
signal: AbortSignal,
|
||||
): Promise<void> {
|
||||
try {
|
||||
await this.client.sendMessage(target.chatId, target.text, signal);
|
||||
} catch (error: unknown) {
|
||||
await this.onSendError(delivery, target.chatId, error);
|
||||
return;
|
||||
}
|
||||
this.lastSendToChat.set(String(target.chatId), performance.now());
|
||||
await settleSuccess(
|
||||
this.prisma,
|
||||
TABLE,
|
||||
delivery.id,
|
||||
delivery.claimedAt,
|
||||
'DELIVERED',
|
||||
);
|
||||
}
|
||||
|
||||
private async onSendError(
|
||||
delivery: ClaimedDelivery,
|
||||
chatId: bigint,
|
||||
error: unknown,
|
||||
): Promise<void> {
|
||||
if (!(error instanceof TelegramApiError)) {
|
||||
await this.failSafely(delivery, error);
|
||||
return;
|
||||
}
|
||||
if (error.statusCode === 429) {
|
||||
const seconds = error.retryAfterSec ?? DEFAULT_RETRY_AFTER_SEC;
|
||||
// Telegram limits the whole bot: the pod waits, not only this row.
|
||||
this.pausedUntil = Math.max(
|
||||
this.pausedUntil,
|
||||
performance.now() + seconds * 1_000,
|
||||
);
|
||||
await deferClaim(
|
||||
this.prisma,
|
||||
TABLE,
|
||||
delivery.id,
|
||||
delivery.claimedAt,
|
||||
seconds * 1_000,
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (isChatGone(error)) {
|
||||
await this.endDelivery(delivery, chatId);
|
||||
return;
|
||||
}
|
||||
if (error.statusCode === 401) {
|
||||
await this.pauseForInvalidToken(delivery);
|
||||
return;
|
||||
}
|
||||
await this.failSafely(delivery, error);
|
||||
}
|
||||
|
||||
/**
|
||||
* The bot is blocked or the chat is gone: the link ends, but only if it
|
||||
* still points to this chat - a newer link of the same user to another
|
||||
* chat stays. Settings, link and message to the user change together.
|
||||
*/
|
||||
private async endDelivery(
|
||||
delivery: ClaimedDelivery,
|
||||
chatId: bigint,
|
||||
): Promise<void> {
|
||||
const userId = delivery.userId;
|
||||
if (userId === null) {
|
||||
await settleSkipped(
|
||||
this.prisma,
|
||||
TABLE,
|
||||
delivery.id,
|
||||
delivery.claimedAt,
|
||||
SKIP.chatUnavailable,
|
||||
);
|
||||
return;
|
||||
}
|
||||
await this.prisma.$transaction(async (tx) => {
|
||||
await lockRowForUpdate(tx, 'NotificationPreference', userId);
|
||||
const { count } = await tx.telegramLink.deleteMany({
|
||||
where: { userId, chatId },
|
||||
});
|
||||
if (count === 1) {
|
||||
await tx.notificationPreference.updateMany({
|
||||
where: { userId },
|
||||
data: { telegramEnabled: false },
|
||||
});
|
||||
await this.notifications.record(
|
||||
{
|
||||
userId,
|
||||
priority: 'HINWEIS',
|
||||
category: 'SYSTEM',
|
||||
message: CHAT_GONE_MESSAGE,
|
||||
},
|
||||
tx,
|
||||
);
|
||||
}
|
||||
await settleSkipped(
|
||||
tx,
|
||||
TABLE,
|
||||
delivery.id,
|
||||
delivery.claimedAt,
|
||||
SKIP.chatUnavailable,
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
/** The token is rejected: nothing can go out, so the whole loop waits. */
|
||||
private async pauseForInvalidToken(delivery: ClaimedDelivery): Promise<void> {
|
||||
this.pausedUntil = performance.now() + TELEGRAM_TOKEN_INVALID_PAUSE_MS;
|
||||
await deferClaim(
|
||||
this.prisma,
|
||||
TABLE,
|
||||
delivery.id,
|
||||
delivery.claimedAt,
|
||||
TELEGRAM_TOKEN_INVALID_PAUSE_MS,
|
||||
);
|
||||
try {
|
||||
// One per user and day: the dedupe key carries the Berlin day.
|
||||
await this.notifications.recordForPermission('system:manage', {
|
||||
category: 'SYSTEM',
|
||||
priority: 'WARNUNG',
|
||||
message: TOKEN_INVALID_MESSAGE,
|
||||
link: '/settings/system',
|
||||
dedupeKeyPrefix: `telegram-token-invalid:${berlinToday()}`,
|
||||
});
|
||||
} catch (error: unknown) {
|
||||
this.logger.error(
|
||||
`Telegram token notice not recorded: ${loggableError(error).message}`,
|
||||
);
|
||||
}
|
||||
this.logger.error(
|
||||
'Telegram rejected the bot token (401); sending paused for 15 minutes',
|
||||
);
|
||||
}
|
||||
|
||||
/** A failed attempt under the shared retry rule; the error text carries no secret. */
|
||||
private async failSafely(
|
||||
delivery: ClaimedDelivery,
|
||||
error: unknown,
|
||||
): Promise<void> {
|
||||
const { message } = loggableError(error);
|
||||
try {
|
||||
await settleFailure(
|
||||
this.prisma,
|
||||
TABLE,
|
||||
delivery.id,
|
||||
delivery.claimedAt,
|
||||
delivery.attempts,
|
||||
message,
|
||||
);
|
||||
} catch (settleError: unknown) {
|
||||
this.logger.error(
|
||||
`Telegram delivery ${delivery.id} not settled: ${loggableError(settleError).message}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Only development may fall back to localhost. Elsewhere a missing origin
|
||||
* (the startup check normally prevents it) sends the text without a link.
|
||||
*/
|
||||
private frontendOrigin(): string | null {
|
||||
const configured = process.env.FRONTEND_ORIGIN;
|
||||
if (configured) {
|
||||
return configured;
|
||||
}
|
||||
const environment = process.env.NODE_ENV;
|
||||
if (environment === undefined || environment === 'development') {
|
||||
return DEVELOPMENT_ORIGIN;
|
||||
}
|
||||
if (!this.warnedMissingOrigin) {
|
||||
this.warnedMissingOrigin = true;
|
||||
this.logger.warn(
|
||||
'FRONTEND_ORIGIN is not set: Telegram messages go out without a link',
|
||||
);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
private remainingPauseMs(): number {
|
||||
return Math.max(0, Math.ceil(this.pausedUntil - performance.now()));
|
||||
}
|
||||
|
||||
/** Milliseconds until the chat may get its next message; 0: now. */
|
||||
private chatWaitMs(chatId: bigint): number {
|
||||
const now = performance.now();
|
||||
for (const [chat, sentAt] of this.lastSendToChat) {
|
||||
if (now - sentAt > CHAT_MEMORY_MS) {
|
||||
this.lastSendToChat.delete(chat);
|
||||
}
|
||||
}
|
||||
const sentAt = this.lastSendToChat.get(String(chatId));
|
||||
return sentAt === undefined
|
||||
? 0
|
||||
: Math.max(0, Math.ceil(TELEGRAM_CHAT_SPACING_MS - (now - sentAt)));
|
||||
}
|
||||
|
||||
/** At most TELEGRAM_MAX_PER_SECOND messages per second over all chats. */
|
||||
private async keepRateLimit(signal: AbortSignal): Promise<void> {
|
||||
const waitMs = MIN_SEND_GAP_MS - (performance.now() - this.lastSendAt);
|
||||
if (waitMs > 0) {
|
||||
await sleep(waitMs, signal);
|
||||
}
|
||||
this.lastSendAt = performance.now();
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user