import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest';
import { PrismaClient } from '@prisma/client';
import type { CloudEvent } from '@insignia/iios-contracts';
import { resetDb } from '../test-utils/reset-db';
import { ProjectionCursorService } from './projection-cursor.service';
import type { PrismaService } from '../prisma/prisma.service';

const url = process.env.DATABASE_URL ?? 'postgresql://iios:iios@localhost:5434/iios?schema=public';
const prisma = new PrismaClient({ datasources: { db: { url } } });
const asService = prisma as unknown as PrismaService;
const svc = () => new ProjectionCursorService(asService);

const TOPIC = 'com.insignia.iios.test.v1';
const ce = (id: string, data: Record<string, unknown> = {}, scope = 'scope-1', type = TOPIC): CloudEvent =>
  ({
    specversion: '1.0',
    id,
    type,
    source: 'iios/test',
    time: '2026-01-01T00:00:00.000Z',
    insignia: { scopeSnapshotId: scope, idempotencyKey: id, dataClass: 'internal' },
    data,
  }) as unknown as CloudEvent;

beforeAll(async () => { await prisma.$connect(); });
afterAll(async () => { await prisma.$disconnect(); });
beforeEach(async () => { await resetDb(prisma); });

describe('ProjectionCursorService (P9 — replay cursor, KG-06)', () => {
  it('advance creates a cursor at offset 1 with the event id and a 64-hex checksum', async () => {
    const row = await svc().advance('p', ce('e1', { n: 1 }));
    expect(row.lastOffset).toBe(1);
    expect(row.lastEventId).toBe('e1');
    expect(row.checksum).toMatch(/^[0-9a-f]{64}$/);
  });

  it('three distinct events advance the offset to 3 and track the latest event id', async () => {
    const s = svc();
    await s.advance('p', ce('e1'));
    await s.advance('p', ce('e2'));
    const third = await s.advance('p', ce('e3'));
    expect(third.lastOffset).toBe(3);
    expect(third.lastEventId).toBe('e3');
  });

  it('replay determinism: re-running the same ordered stream reproduces the checksum (KG-06)', async () => {
    const s = svc();
    await s.advance('p', ce('e1', { a: 1 }));
    await s.advance('p', ce('e2', { b: 2 }));
    const c1 = (await s.advance('p', ce('e3', { c: 3 }))).checksum;

    // Wipe the cursor and replay the identical stream.
    await prisma.iiosProjectionCursor.deleteMany({});
    await s.advance('p', ce('e1', { a: 1 }));
    await s.advance('p', ce('e2', { b: 2 }));
    const c2 = (await s.advance('p', ce('e3', { c: 3 }))).checksum;

    expect(c2).toBe(c1);
  });

  it('order sensitivity: a different event order yields a different checksum', async () => {
    const s = svc();
    await s.advance('p', ce('e1'));
    const ab = (await s.advance('p', ce('e2'))).checksum;

    await prisma.iiosProjectionCursor.deleteMany({});
    await s.advance('p', ce('e2'));
    const ba = (await s.advance('p', ce('e1'))).checksum;

    expect(ba).not.toBe(ab);
  });

  it('a different source_topic for the same projection is a separate cursor row', async () => {
    const s = svc();
    await s.advance('p', ce('e1', {}, 'scope-1', 'topic.a'));
    await s.advance('p', ce('e2', {}, 'scope-1', 'topic.b'));
    expect(await prisma.iiosProjectionCursor.count({ where: { projectionName: 'p' } })).toBe(2);
  });

  it('two partitions (scopes) track independently', async () => {
    const s = svc();
    const a = await s.advance('p', ce('e1', {}, 'scope-A'));
    const b = await s.advance('p', ce('e2', {}, 'scope-B'));
    expect(a.partitionKey).toBe('scope-A');
    expect(b.partitionKey).toBe('scope-B');
    expect(await prisma.iiosProjectionCursor.count()).toBe(2);
  });
});
