import { createHash, createHmac } from 'node:crypto';
import { Inject, Injectable } from '@nestjs/common';
import {
  and,
  asc,
  desc,
  eq,
  interactionCheckpoints,
  interactionHandoffs,
  interactionInbox,
  interactionOutbox,
  interactionSessions,
  type Db,
} from '@mosaicstack/db';
import { seal, unseal } from '@mosaicstack/auth';
import { redactSensitiveContent } from '@mosaicstack/log';
import type {
  DurableCheckpoint,
  DurableCheckpointInput,
  DurableEnqueueResult,
  DurableHandoff,
  DurableHandoffInput,
  DurableInboxEntry,
  DurableInboxInput,
  DurableInboxStatus,
  DurableOutboxEntry,
  DurableOutboxInput,
  DurableOutboxStatus,
  DurableSessionIdentity,
  DurableSessionSnapshot,
  DurableSessionStore,
} from '@mosaicstack/agent';
import { DB } from '../database/database.module.js';

@Injectable()
export class TessDurableSessionRepository implements DurableSessionStore {
  constructor(@Inject(DB) private readonly db: Db) {}

  async create(identity: DurableSessionIdentity): Promise<void> {
    await this.db
      .insert(interactionSessions)
      .values({
        id: identity.sessionId,
        agentName: identity.agentName,
        tenantId: identity.tenantId,
        ownerId: identity.ownerId,
        providerId: identity.providerId,
        runtimeSessionId: identity.runtimeSessionId,
      })
      .onConflictDoNothing();

    const existing = await this.session(identity.sessionId);
    if (!existing || !sameEnrollmentScope(existing, identity)) {
      throw new Error(`Durable Tess session identity conflict: ${identity.sessionId}`);
    }
    // A recovered/re-enrolled runtime can receive a new provider session ID;
    // the conversation handle and owner scope remain immutable.
    if (
      existing.providerId !== identity.providerId ||
      existing.runtimeSessionId !== identity.runtimeSessionId
    ) {
      await this.db
        .update(interactionSessions)
        .set({ providerId: identity.providerId, runtimeSessionId: identity.runtimeSessionId })
        .where(eq(interactionSessions.id, identity.sessionId));
    }
  }

  async snapshot(sessionId: string): Promise<DurableSessionSnapshot | null> {
    const identity = await this.session(sessionId);
    if (!identity) return null;

    const [inbox, outbox, checkpoints, handoffs] = await Promise.all([
      this.db
        .select()
        .from(interactionInbox)
        .where(eq(interactionInbox.sessionId, sessionId))
        .orderBy(asc(interactionInbox.createdAt)),
      this.db
        .select()
        .from(interactionOutbox)
        .where(eq(interactionOutbox.sessionId, sessionId))
        .orderBy(asc(interactionOutbox.createdAt)),
      this.db
        .select()
        .from(interactionCheckpoints)
        .where(eq(interactionCheckpoints.sessionId, sessionId))
        .orderBy(
          desc(interactionCheckpoints.compactionEpoch),
          desc(interactionCheckpoints.createdAt),
        )
        .limit(1),
      this.db
        .select()
        .from(interactionHandoffs)
        .where(eq(interactionHandoffs.sessionId, sessionId))
        .orderBy(asc(interactionHandoffs.createdAt)),
    ]);

    const checkpoint = checkpoints[0];
    return {
      identity,
      inbox: inbox.map(toInbox),
      outbox: outbox.map(toOutbox),
      ...(checkpoint ? { checkpoint: toCheckpoint(checkpoint) } : {}),
      handoffs: handoffs.map(toHandoff),
    };
  }

  async enqueueInbox(input: DurableInboxInput): Promise<DurableEnqueueResult<DurableInboxStatus>> {
    const digest = contentDigest(input.content);
    const record: DurableInboxInput = {
      ...input,
      content: redactSensitiveContent(input.content).content,
    };
    const inserted = await this.db
      .insert(interactionInbox)
      .values({
        ...record,
        content: seal(record.content),
        contentDigest: digest,
        status: 'pending',
      })
      .onConflictDoNothing()
      .returning({ status: interactionInbox.status });
    if (inserted[0]) return { accepted: true, status: inserted[0].status };

    const existing = await this.db
      .select()
      .from(interactionInbox)
      .where(
        and(
          eq(interactionInbox.sessionId, input.sessionId),
          eq(interactionInbox.idempotencyKey, input.idempotencyKey),
        ),
      )
      .limit(1);
    if (!existing[0]) throw new Error(`Durable Tess inbox enqueue failed: ${input.idempotencyKey}`);
    const entry = toInbox(existing[0]);
    if (
      !sameInbox(entry, record) ||
      !matchesContentDigest(existing[0].contentDigest, input.content)
    ) {
      throw new Error(`Durable Tess inbox idempotency conflict: ${input.idempotencyKey}`);
    }
    return { accepted: false, status: entry.status };
  }

  async claimInbox(sessionId: string): Promise<DurableInboxEntry | null> {
    for (let attempt = 0; attempt < 3; attempt += 1) {
      const candidate = await this.db
        .select()
        .from(interactionInbox)
        .where(
          and(eq(interactionInbox.sessionId, sessionId), eq(interactionInbox.status, 'pending')),
        )
        .orderBy(asc(interactionInbox.createdAt))
        .limit(1);
      const entry = candidate[0];
      if (!entry) return null;
      const claimed = await this.db
        .update(interactionInbox)
        .set({ status: 'processing', updatedAt: new Date() })
        .where(and(eq(interactionInbox.id, entry.id), eq(interactionInbox.status, 'pending')))
        .returning();
      if (claimed[0]) return toInbox(claimed[0]);
    }
    return null;
  }

  async completeInbox(sessionId: string, idempotencyKey: string): Promise<void> {
    await this.db
      .update(interactionInbox)
      .set({ status: 'processed', updatedAt: new Date() })
      .where(
        and(
          eq(interactionInbox.sessionId, sessionId),
          eq(interactionInbox.idempotencyKey, idempotencyKey),
          eq(interactionInbox.status, 'processing'),
        ),
      );
  }

  async releaseInbox(sessionId: string, idempotencyKey: string): Promise<void> {
    await this.db
      .update(interactionInbox)
      .set({ status: 'pending', updatedAt: new Date() })
      .where(
        and(
          eq(interactionInbox.sessionId, sessionId),
          eq(interactionInbox.idempotencyKey, idempotencyKey),
          eq(interactionInbox.status, 'processing'),
        ),
      );
  }

  async enqueueOutbox(
    input: DurableOutboxInput,
  ): Promise<DurableEnqueueResult<DurableOutboxStatus>> {
    const digest = contentDigest(input.content);
    const record: DurableOutboxInput = {
      ...input,
      content: redactSensitiveContent(input.content).content,
    };
    const inserted = await this.db
      .insert(interactionOutbox)
      .values({
        ...record,
        content: seal(record.content),
        contentDigest: digest,
        status: 'pending',
      })
      .onConflictDoNothing()
      .returning({ status: interactionOutbox.status });
    if (inserted[0]) return { accepted: true, status: inserted[0].status };

    const existing = await this.db
      .select()
      .from(interactionOutbox)
      .where(
        and(
          eq(interactionOutbox.sessionId, input.sessionId),
          eq(interactionOutbox.idempotencyKey, input.idempotencyKey),
        ),
      )
      .limit(1);
    if (!existing[0])
      throw new Error(`Durable Tess outbox enqueue failed: ${input.idempotencyKey}`);
    const entry = toOutbox(existing[0]);
    if (
      !sameOutbox(entry, record) ||
      !matchesContentDigest(existing[0].contentDigest, input.content)
    ) {
      throw new Error(`Durable Tess outbox idempotency conflict: ${input.idempotencyKey}`);
    }
    return { accepted: false, status: entry.status };
  }

  async claimOutbox(sessionId: string): Promise<DurableOutboxEntry | null> {
    for (let attempt = 0; attempt < 3; attempt += 1) {
      const candidate = await this.db
        .select()
        .from(interactionOutbox)
        .where(
          and(eq(interactionOutbox.sessionId, sessionId), eq(interactionOutbox.status, 'pending')),
        )
        .orderBy(asc(interactionOutbox.createdAt))
        .limit(1);
      const entry = candidate[0];
      if (!entry) return null;
      const claimed = await this.db
        .update(interactionOutbox)
        .set({ status: 'processing', updatedAt: new Date() })
        .where(and(eq(interactionOutbox.id, entry.id), eq(interactionOutbox.status, 'pending')))
        .returning();
      if (claimed[0]) return toOutbox(claimed[0]);
    }
    return null;
  }

  async claimOutboxByKey(
    sessionId: string,
    idempotencyKey: string,
  ): Promise<DurableOutboxEntry | null> {
    const claimed = await this.db
      .update(interactionOutbox)
      .set({ status: 'processing', updatedAt: new Date() })
      .where(
        and(
          eq(interactionOutbox.sessionId, sessionId),
          eq(interactionOutbox.idempotencyKey, idempotencyKey),
          eq(interactionOutbox.status, 'pending'),
        ),
      )
      .returning();
    return claimed[0] ? toOutbox(claimed[0]) : null;
  }

  async completeOutbox(sessionId: string, idempotencyKey: string): Promise<void> {
    await this.db
      .update(interactionOutbox)
      .set({ status: 'delivered', updatedAt: new Date() })
      .where(
        and(
          eq(interactionOutbox.sessionId, sessionId),
          eq(interactionOutbox.idempotencyKey, idempotencyKey),
          eq(interactionOutbox.status, 'processing'),
        ),
      );
  }

  async releaseOutbox(sessionId: string, idempotencyKey: string): Promise<void> {
    await this.db
      .update(interactionOutbox)
      .set({ status: 'pending', updatedAt: new Date() })
      .where(
        and(
          eq(interactionOutbox.sessionId, sessionId),
          eq(interactionOutbox.idempotencyKey, idempotencyKey),
          eq(interactionOutbox.status, 'processing'),
        ),
      );
  }

  async checkpoint(input: DurableCheckpointInput): Promise<void> {
    // Compute identity before redaction.  The persisted digest is keyed so a database
    // reader cannot use it as an offline oracle for sensitive cursor/summary values.
    const digest = contentDigest(JSON.stringify([input.cursor, input.summary]));
    const checkpoint: DurableCheckpointInput = {
      ...input,
      cursor: redactSensitiveContent(input.cursor).content,
      summary: redactSensitiveContent(input.summary).content,
    };
    const inserted = await this.db
      .insert(interactionCheckpoints)
      .values({
        ...checkpoint,
        contentDigest: digest,
        cursor: seal(checkpoint.cursor),
        summary: seal(checkpoint.summary),
      })
      .onConflictDoNothing()
      .returning({ checkpointId: interactionCheckpoints.checkpointId });
    if (inserted[0]) return;

    const existing = await this.db
      .select()
      .from(interactionCheckpoints)
      .where(
        and(
          eq(interactionCheckpoints.sessionId, input.sessionId),
          eq(interactionCheckpoints.checkpointId, input.checkpointId),
        ),
      )
      .limit(1);
    if (
      !existing[0] ||
      !sameCheckpoint(toCheckpoint(existing[0]), checkpoint) ||
      !matchesCheckpointDigest(existing[0].contentDigest, digest)
    ) {
      throw new Error(`Durable Tess checkpoint identity conflict: ${input.checkpointId}`);
    }
  }

  async findCheckpoint(sessionId: string, checkpointId: string): Promise<DurableCheckpoint | null> {
    const checkpoints = await this.db
      .select()
      .from(interactionCheckpoints)
      .where(
        and(
          eq(interactionCheckpoints.sessionId, sessionId),
          eq(interactionCheckpoints.checkpointId, checkpointId),
        ),
      )
      .limit(1);
    const checkpoint = checkpoints[0];
    return checkpoint ? toCheckpoint(checkpoint) : null;
  }

  async handoff(input: DurableHandoffInput): Promise<void> {
    const checkpoint = await this.findCheckpoint(input.sessionId, input.checkpointId);
    if (!checkpoint) {
      throw new Error(`Durable Tess handoff checkpoint is unavailable: ${input.checkpointId}`);
    }
    const inserted = await this.db
      .insert(interactionHandoffs)
      .values({ ...input })
      .onConflictDoNothing()
      .returning({ handoffId: interactionHandoffs.handoffId });
    if (inserted[0]) return;

    const existing = await this.findHandoff(input.handoffId);
    if (!existing || !sameHandoff(existing, input)) {
      throw new Error(`Durable Tess handoff identity conflict: ${input.handoffId}`);
    }
  }

  async findHandoff(handoffId: string): Promise<DurableHandoff | null> {
    const handoffs = await this.db
      .select()
      .from(interactionHandoffs)
      .where(eq(interactionHandoffs.handoffId, handoffId))
      .limit(1);
    const handoff = handoffs[0];
    return handoff ? toHandoff(handoff) : null;
  }

  async requeueInFlight(sessionId: string): Promise<void> {
    // Inbox handlers are process-local work. A provider outbox claim may have
    // reached an external target before a crash, so it is deliberately not
    // replayed by generic recovery.
    await this.db
      .update(interactionInbox)
      .set({ status: 'pending', updatedAt: new Date() })
      .where(
        and(eq(interactionInbox.sessionId, sessionId), eq(interactionInbox.status, 'processing')),
      );
  }

  private async session(sessionId: string): Promise<DurableSessionIdentity | null> {
    const sessions = await this.db
      .select()
      .from(interactionSessions)
      .where(eq(interactionSessions.id, sessionId))
      .limit(1);
    const session = sessions[0];
    return session
      ? {
          agentName: session.agentName,
          sessionId: session.id,
          tenantId: session.tenantId,
          ownerId: session.ownerId,
          providerId: session.providerId,
          runtimeSessionId: session.runtimeSessionId,
        }
      : null;
  }
}

function contentDigest(content: string): string {
  const secret = process.env['BETTER_AUTH_SECRET'];
  if (!secret) {
    throw new Error('BETTER_AUTH_SECRET is required for durable idempotency digests');
  }
  return `hmac:v1:${createHmac('sha256', secret).update(content).digest('hex')}`;
}

function matchesContentDigest(stored: string, content: string): boolean {
  return (
    stored === contentDigest(content) ||
    stored === createHash('sha256').update(content).digest('hex')
  );
}

function matchesCheckpointDigest(stored: string, digest: string): boolean {
  // Legacy rows predate any pre-redaction identity and cannot safely prove equality.
  // Reject rather than let redaction collapse distinct sensitive checkpoint payloads.
  return stored === digest;
}

function sameEnrollmentScope(left: DurableSessionIdentity, right: DurableSessionIdentity): boolean {
  return (
    left.agentName === right.agentName &&
    left.sessionId === right.sessionId &&
    left.tenantId === right.tenantId &&
    left.ownerId === right.ownerId
  );
}

function sameInbox(left: DurableInboxEntry, right: DurableInboxInput): boolean {
  return (
    left.sessionId === right.sessionId &&
    left.idempotencyKey === right.idempotencyKey &&
    left.correlationId === right.correlationId &&
    left.content === right.content
  );
}

function sameOutbox(left: DurableOutboxEntry, right: DurableOutboxInput): boolean {
  return (
    left.sessionId === right.sessionId &&
    left.idempotencyKey === right.idempotencyKey &&
    left.correlationId === right.correlationId &&
    left.channelId === right.channelId &&
    left.kind === right.kind &&
    left.content === right.content
  );
}

function sameCheckpoint(left: DurableCheckpoint, right: DurableCheckpointInput): boolean {
  return (
    left.sessionId === right.sessionId &&
    left.checkpointId === right.checkpointId &&
    left.compactionEpoch === right.compactionEpoch
  );
}

function sameHandoff(left: DurableHandoff, right: DurableHandoffInput): boolean {
  return (
    left.sessionId === right.sessionId &&
    left.handoffId === right.handoffId &&
    left.destination === right.destination &&
    left.correlationId === right.correlationId &&
    left.checkpointId === right.checkpointId &&
    left.status === right.status
  );
}

function toInbox(row: typeof interactionInbox.$inferSelect): DurableInboxEntry {
  return {
    sessionId: row.sessionId,
    idempotencyKey: row.idempotencyKey,
    correlationId: row.correlationId,
    content: unseal(row.content),
    status: row.status,
  };
}

function toOutbox(row: typeof interactionOutbox.$inferSelect): DurableOutboxEntry {
  return {
    sessionId: row.sessionId,
    idempotencyKey: row.idempotencyKey,
    correlationId: row.correlationId,
    channelId: row.channelId,
    kind: row.kind,
    content: unseal(row.content),
    status: row.status,
  };
}

function toCheckpoint(row: typeof interactionCheckpoints.$inferSelect): DurableCheckpoint {
  return {
    sessionId: row.sessionId,
    checkpointId: row.checkpointId,
    cursor: unseal(row.cursor),
    summary: unseal(row.summary),
    compactionEpoch: row.compactionEpoch,
  };
}

function toHandoff(row: typeof interactionHandoffs.$inferSelect): DurableHandoff {
  return {
    sessionId: row.sessionId,
    handoffId: row.handoffId,
    destination: row.destination,
    correlationId: row.correlationId,
    checkpointId: row.checkpointId,
    status: row.status,
  };
}