import { err, ok, type CommandContext } from "@tailor-platform/erp-kit/core";
import type { Transaction } from "../generated/kysely-tailordb";
import {
  InvalidPayloadError,
  MissingEntityReferenceError,
  MissingRequiredFieldError,
} from "../lib/errors.generated";
import { hashPayload, normalizePayload, type NotificationEventPayload } from "../lib/events";

export interface LogNotificationEventInput {
  eventType: string;
  sourceType: string;
  sourceId: string;
  actorUserId?: string | null;
  payload: string | NotificationEventPayload;
}

/**
 * Append a source domain event into NotificationEvent. The command is the
 * stable, domain-agnostic ingress seam: it records the event verbatim and
 * leaves all delivery work to the CDC-triggered dispatch executor. It never
 * reads source-module tables and does not validate `eventType` against a
 * module-owned vocabulary — catalog resolution happens later, at dispatch.
 *
 * Idempotent on `(eventType, sourceType, sourceId, payloadHash)`.
 */
export async function run(db: Transaction, input: LogNotificationEventInput, ctx: CommandContext) {
  void ctx;

  if (!input.eventType) return err(new MissingRequiredFieldError("eventType"));
  if (!input.sourceType || !input.sourceId)
    return err(new MissingEntityReferenceError(`${input.sourceType}:${input.sourceId}`));

  const { raw, parsed } = normalizePayload(input.payload);
  if (!raw || !parsed) return err(new InvalidPayloadError(JSON.stringify(input.payload)));

  const payloadHash = await hashPayload(raw);
  const existing = await db
    .selectFrom("NotificationEvent")
    .selectAll()
    .where("eventType", "=", input.eventType)
    .where("sourceType", "=", input.sourceType)
    .where("sourceId", "=", input.sourceId)
    .where("payloadHash", "=", payloadHash)
    .executeTakeFirst();

  if (existing) return ok({ notificationEvent: existing, duplicate: true });

  const now = new Date();
  try {
    const notificationEvent = await db
      .insertInto("NotificationEvent")
      .values({
        eventType: input.eventType,
        sourceType: input.sourceType,
        sourceId: input.sourceId,
        actorUserId: input.actorUserId ?? null,
        payload: raw,
        payloadHash,
        status: "PENDING",
        dispatchedAt: null,
        createdAt: now,
      })
      .returningAll()
      .executeTakeFirst();

    return ok({ notificationEvent, duplicate: false });
  } catch (thrown) {
    // Concurrent emit: another transaction may have inserted the same tuple
    // between the idempotency probe and our insert, tripping the unique index
    // on (eventType, sourceType, sourceId, payloadHash). Re-select the tuple
    // and fold into the duplicate path; anything else (or a phantom failure
    // with no row) rethrows so the emitter's transaction is not corrupted.
    const raced = await db
      .selectFrom("NotificationEvent")
      .selectAll()
      .where("eventType", "=", input.eventType)
      .where("sourceType", "=", input.sourceType)
      .where("sourceId", "=", input.sourceId)
      .where("payloadHash", "=", payloadHash)
      .executeTakeFirst();

    if (raced) return ok({ notificationEvent: raced, duplicate: true });
    // Not a duplicate race: rethrow (rather than err) so the emitter's
    // surrounding transaction aborts instead of committing a half-applied
    // business change without its event row.
    // oxlint-disable-next-line eslint-js/no-restricted-syntax
    throw thrown;
  }
}