import { Injectable } from '@nestjs/common';
import { createHash } from 'node:crypto';
import type { CloudEvent } from '@insignia/iios-contracts';
import { PrismaService } from '../prisma/prisma.service';

const sha = (s: string) => createHash('sha256').update(s).digest('hex');
// Stable stringify: sort object keys so {a,b} and {b,a} hash equal.
const stable = (v: unknown): string =>
  v === null || typeof v !== 'object'
    ? JSON.stringify(v) ?? 'null'
    : Array.isArray(v)
      ? `[${v.map(stable).join(',')}]`
      : `{${Object.keys(v as object)
          .sort()
          .map((k) => `${JSON.stringify(k)}:${stable((v as Record<string, unknown>)[k])}`)
          .join(',')}}`;

/**
 * Projection replay cursor (P9, KG-06). Each projection worker folds every event it
 * consumes into an ordered high-water-mark (lastEventId + monotonic lastOffset) and a
 * rolling checksum, keyed per (projection, source_topic, partition). Because the checksum
 * is a deterministic function of the ordered (eventId, eventHash) stream, replaying the
 * same events reproduces the same checksum — the proof that replay reproduces outcomes.
 */
@Injectable()
export class ProjectionCursorService {
  constructor(private readonly prisma: PrismaService) {}

  /** Fold one consumed event into the projection's ordered cursor + rolling checksum. */
  async advance(projectionName: string, event: CloudEvent) {
    const sourceTopic = event.type;
    const partitionKey = event.insignia?.scopeSnapshotId ?? 'global';
    const eventHash = sha(stable({ id: event.id, type: event.type, data: event.data }));
    const key = { projectionName_sourceTopic_partitionKey: { projectionName, sourceTopic, partitionKey } };

    const prev = await this.prisma.iiosProjectionCursor.findUnique({ where: key });
    const checksum = sha(`${prev?.checksum ?? ''}${event.id}${eventHash}`);
    const lastOffset = (prev?.lastOffset ?? 0) + 1;

    return this.prisma.iiosProjectionCursor.upsert({
      where: key,
      create: { projectionName, sourceTopic, partitionKey, lastEventId: event.id, lastOffset, checksum },
      update: { lastEventId: event.id, lastOffset, checksum },
    });
  }

  /** All cursors (global ops read for /metrics + replay/repair tooling). */
  async list() {
    return this.prisma.iiosProjectionCursor.findMany({ orderBy: [{ projectionName: 'asc' }, { sourceTopic: 'asc' }] });
  }
}
