ApiCore/ApiCore-Backend/test/telegram-schema.e2e-spec.ts

import { Prisma } from '../generated/prisma';
import {
  deferClaim,
  claimBatch,
  releaseStaleClaims,
  settleSkipped,
  settleSuccess,
} from '../src/common/outbox/outbox-claims';
import { PrismaService } from '../src/prisma/prisma.service';

interface Claimed {
  id: string;
  claimedAt: Date;
}

/**
 * Schema-level guarantees of the T4 Telegram migration and the outbox
 * helpers on NotificationDelivery: the target CHECK, the poll-state
 * singleton, cascade with the user, compare-and-set settling.
 */
describe('Telegram schema and delivery outbox (e2e)', () => {
  let prisma: PrismaService;
  const suffix = `${Date.now()}-${Math.round(Math.random() * 1e6)}`;
  let userId: string;
  let notificationId: string;
  const chatBase = BigInt(9_000_000_000) + BigInt(Date.now() % 1_000_000);

  beforeAll(async () => {
    prisma = new PrismaService();
    const user = await prisma.user.create({
      data: {
        email: `tg-schema-${suffix}@example.com`,
        passwordHash: 'x',
        name: 'Telegram Schema',
        role: { connect: { key: 'READONLY' } },
      },
    });
    userId = user.id;
    const notification = await prisma.notification.create({
      data: {
        userId,
        priority: 'INFO',
        category: 'SYSTEM',
        message: 'm',
      },
    });
    notificationId = notification.id;
  });

  afterAll(async () => {
    await prisma.user.deleteMany({ where: { id: userId } });
    await prisma.$disconnect();
  });

  afterEach(async () => {
    await prisma.notificationDelivery.deleteMany({
      where: { OR: [{ userId }, { chatId: { gte: chatBase } }] },
    });
  });

  it('has exactly one TelegramPollState row, created by the migration', async () => {
    const rows = await prisma.telegramPollState.findMany();
    expect(rows).toHaveLength(1);
    expect(rows[0]).toMatchObject({ id: 'singleton', updateOffset: 0n });
  });

  describe('NotificationDelivery target CHECK', () => {
    const base = { channel: 'TELEGRAM' } as const;

    it('accepts a notification for a user', async () => {
      await expect(
        prisma.notificationDelivery.create({
          data: { ...base, userId, notificationId },
        }),
      ).resolves.toMatchObject({ status: 'PENDING', attempts: 0 });
    });

    it('accepts a system reply for a chat', async () => {
      await expect(
        prisma.notificationDelivery.create({
          data: { ...base, chatId: chatBase, templateCode: 'HELP' },
        }),
      ).resolves.toMatchObject({ status: 'PENDING' });
    });

    it.each([
      ['an empty row', {}],
      ['notification without user', { notificationId: 'n' }],
      ['template without chat', { templateCode: 'HELP' }],
      ['chat without template', { chatId: chatBase }],
      [
        'notification and template',
        {
          userId: 'u',
          notificationId: 'n',
          templateCode: 'HELP',
          chatId: chatBase,
        },
      ],
    ])('rejects %s', async (_name, extra) => {
      const insert = prisma.$executeRaw`INSERT INTO "NotificationDelivery" ("id","channel","userId","chatId","notificationId","templateCode")
        VALUES (${`chk-${suffix}-${Math.random()}`}, 'TELEGRAM',
          ${(extra as { userId?: string }).userId ?? null},
          ${(extra as { chatId?: bigint }).chatId ?? null},
          ${(extra as { notificationId?: string }).notificationId ?? null},
          ${(extra as { templateCode?: string }).templateCode ?? null})`;
      await expect(insert).rejects.toThrow();
    });
  });

  it('allows only the singleton row in TelegramPollState', async () => {
    await expect(
      prisma.$executeRaw`INSERT INTO "TelegramPollState" ("id") VALUES (${`other-${suffix}`})`,
    ).rejects.toThrow();
  });

  it('creates at most one delivery per notification and channel', async () => {
    await prisma.notificationDelivery.create({
      data: { channel: 'TELEGRAM', userId, notificationId },
    });

    await expect(
      prisma.notificationDelivery.create({
        data: { channel: 'TELEGRAM', userId, notificationId },
      }),
    ).rejects.toThrow();
  });

  it('indexes NotificationDelivery by notificationId', async () => {
    const rows = await prisma.$queryRaw<{ indexdef: string }[]>`
      SELECT indexdef FROM pg_indexes
      WHERE tablename = 'NotificationDelivery' AND indexdef LIKE '%("notificationId")%'`;
    expect(rows.length).toBeGreaterThanOrEqual(1);
  });

  describe('outbox helpers', () => {
    async function claimOne(): Promise<Claimed> {
      const created = await prisma.notificationDelivery.create({
        data: { channel: 'TELEGRAM', userId, notificationId },
      });
      const [claimed] = await claimBatch<Claimed>(
        prisma,
        'NotificationDelivery',
        {
          limit: 1,
          filter: Prisma.sql`"id" = ${created.id}`,
          orderBy: 'createdAt',
        },
      );
      expect(claimed.id).toBe(created.id);
      return claimed;
    }

    it('settleSkipped needs the matching claim and records the reason', async () => {
      const claimed = await claimOne();

      await expect(
        settleSkipped(
          prisma,
          'NotificationDelivery',
          claimed.id,
          new Date(claimed.claimedAt.getTime() + 1),
          'nope',
        ),
      ).resolves.toBe(false);
      await expect(
        settleSkipped(
          prisma,
          'NotificationDelivery',
          claimed.id,
          claimed.claimedAt,
          'chat gone',
        ),
      ).resolves.toBe(true);

      await expect(
        prisma.notificationDelivery.findUnique({ where: { id: claimed.id } }),
      ).resolves.toMatchObject({
        status: 'SKIPPED',
        lastError: 'chat gone',
        claimedAt: null,
        attempts: 0,
      });
    });

    it('deferClaim puts the entry back for later without counting an attempt', async () => {
      const claimed = await claimOne();

      await expect(
        deferClaim(
          prisma,
          'NotificationDelivery',
          claimed.id,
          new Date(0),
          1000,
        ),
      ).resolves.toBe(false);
      await expect(
        deferClaim(
          prisma,
          'NotificationDelivery',
          claimed.id,
          claimed.claimedAt,
          60_000,
        ),
      ).resolves.toBe(true);

      const row = await prisma.notificationDelivery.findUniqueOrThrow({
        where: { id: claimed.id },
      });
      expect(row).toMatchObject({
        status: 'PENDING',
        attempts: 0,
        claimedAt: null,
      });
      const [{ ahead }] = await prisma.$queryRaw<{ ahead: number }[]>`
        SELECT EXTRACT(EPOCH FROM ("availableAt" - now()))::float8 AS ahead
        FROM "NotificationDelivery" WHERE "id" = ${claimed.id}`;
      expect(ahead).toBeGreaterThan(50);
      expect(ahead).toBeLessThanOrEqual(61);
    });

    it('settleSuccess marks DELIVERED with deliveredAt', async () => {
      const claimed = await claimOne();

      await expect(
        settleSuccess(
          prisma,
          'NotificationDelivery',
          claimed.id,
          claimed.claimedAt,
          'DELIVERED',
        ),
      ).resolves.toBe(true);

      const row = await prisma.notificationDelivery.findUniqueOrThrow({
        where: { id: claimed.id },
      });
      expect(row.status).toBe('DELIVERED');
      expect(row.deliveredAt).not.toBeNull();
    });

    it('the stale release works on NotificationDelivery', async () => {
      const claimed = await claimOne();
      await prisma.$executeRaw`UPDATE "NotificationDelivery" SET "claimedAt" = now() - interval '1 day' WHERE "id" = ${claimed.id}`;

      expect(
        await releaseStaleClaims(prisma, 'NotificationDelivery'),
      ).toBeGreaterThanOrEqual(1);

      await expect(
        prisma.notificationDelivery.findUniqueOrThrow({
          where: { id: claimed.id },
        }),
      ).resolves.toMatchObject({ status: 'PENDING', attempts: 1 });
    });
  });

  it('deleting the user removes the chat binding, tokens and deliveries', async () => {
    const other = await prisma.user.create({
      data: {
        email: `tg-cascade-${suffix}@example.com`,
        passwordHash: 'x',
        name: 'Cascade',
        role: { connect: { key: 'READONLY' } },
      },
    });
    const note = await prisma.notification.create({
      data: {
        userId: other.id,
        priority: 'INFO',
        category: 'SYSTEM',
        message: 'm',
      },
    });
    await prisma.telegramLink.create({
      data: { userId: other.id, chatId: chatBase + 1n },
    });
    await prisma.telegramLinkToken.create({
      data: {
        userId: other.id,
        tokenHash: `hash-${suffix}`,
        expiresAt: new Date(Date.now() + 60_000),
      },
    });
    await prisma.notificationDelivery.create({
      data: { channel: 'TELEGRAM', userId: other.id, notificationId: note.id },
    });

    await prisma.user.delete({ where: { id: other.id } });

    expect(
      await prisma.telegramLink.count({ where: { userId: other.id } }),
    ).toBe(0);
    expect(
      await prisma.telegramLinkToken.count({ where: { userId: other.id } }),
    ).toBe(0);
    expect(
      await prisma.notificationDelivery.count({ where: { userId: other.id } }),
    ).toBe(0);
  });

  it('one chat belongs to exactly one user', async () => {
    const link = await prisma.telegramLink.create({
      data: { userId, chatId: chatBase + 2n },
    });
    const second = await prisma.user.create({
      data: {
        email: `tg-chat-${suffix}@example.com`,
        passwordHash: 'x',
        name: 'Second',
        role: { connect: { key: 'READONLY' } },
      },
    });
    try {
      await expect(
        prisma.telegramLink.create({
          data: { userId: second.id, chatId: link.chatId },
        }),
      ).rejects.toThrow();
    } finally {
      await prisma.telegramLink.deleteMany({ where: { userId } });
      await prisma.user.delete({ where: { id: second.id } });
    }
  });
});