import express from "express";
import { randomUUID, timingSafeEqual } from "node:crypto";
import { createServer } from "node:http";
import { Pool } from "pg";

const config = readConfig();
const pool = new Pool({ connectionString: config.databaseUrl, max: config.databasePoolSize });
const app = express();
const httpServer = createServer(app);
const streamClients = new Set();

app.disable("x-powered-by");
app.use(express.json({ limit: "1mb" }));

app.get("/healthz", asyncRoute(async (_req, res) => {
  await pool.query("select 1");
  res.json({
    ok: true,
    service: "nodedc-notification-core",
    database: "ready",
    internalApiConfigured: Boolean(config.internalAccessToken),
  });
}));

app.post("/internal/notifications/events", requireInternalApi, asyncRoute(async (req, res) => {
  const command = sanitizeEventCommand(req.body);
  const result = await createNotificationEvent(command);
  res.status(result.created ? 201 : 200).json(result);
}));

app.get("/api/notifications", requireInternalApi, asyncRoute(async (req, res) => {
  const subject = getRequestSubject(req);
  const surface = sanitizeSurface(req.query.surface || "hub");
  const limit = sanitizeLimit(req.query.limit);
  const unreadOnly = req.query.unreadOnly === "1" || req.query.unreadOnly === "true";
  const deliveries = await listDeliveries({ subject, surface, limit, unreadOnly });
  res.json({ ok: true, surface, deliveries });
}));

app.post("/api/notifications/:deliveryId/read", requireInternalApi, asyncRoute(async (req, res) => {
  const subject = getRequestSubject(req);
  const deliveryId = sanitizeUuid(req.params.deliveryId, "deliveryId");
  const delivery = await markDeliveryRead({ subject, deliveryId });
  if (!delivery) {
    res.status(404).json({ ok: false, error: "notification_delivery_not_found" });
    return;
  }
  broadcastDeliveryUpdate("notification.delivery.read", delivery);
  res.json({ ok: true, delivery });
}));

app.post("/api/notifications/read-all", requireInternalApi, asyncRoute(async (req, res) => {
  const subject = getRequestSubject(req);
  const surface = sanitizeSurface(req.query.surface || req.body?.surface || "hub");
  const result = await markAllDeliveriesRead({ subject, surface });
  broadcastSubjectUpdate("notification.deliveries.read-all", { subject, surface, count: result.count });
  res.json({ ok: true, surface, count: result.count });
}));

app.get("/api/notifications/stream", requireInternalApi, (req, res) => {
  const subject = getRequestSubject(req);
  const surface = sanitizeSurface(req.query.surface || "hub");
  const client = { id: randomUUID(), res, subject, surface };

  res.setHeader("Content-Type", "text/event-stream");
  res.setHeader("Cache-Control", "no-cache, no-transform");
  res.setHeader("Connection", "keep-alive");
  res.setHeader("X-Accel-Buffering", "no");
  res.flushHeaders?.();
  writeSse(res, "notification.ready", { ok: true, surface });

  const keepAlive = setInterval(() => {
    res.write(": keep-alive\n\n");
  }, 30000);

  streamClients.add(client);
  req.on("close", () => {
    clearInterval(keepAlive);
    streamClients.delete(client);
  });
});

app.use((error, _req, res, _next) => {
  const status = Number(error?.status || 500);
  const message = error instanceof Error ? error.message : "internal_error";
  res.status(status >= 400 && status < 600 ? status : 500).json({ ok: false, error: message });
});

await migrate();

httpServer.listen(config.port, "0.0.0.0", () => {
  console.log(`NODE.DC Notification Core listening on http://0.0.0.0:${config.port}`);
});

process.on("SIGTERM", shutdown);
process.on("SIGINT", shutdown);

async function createNotificationEvent(command) {
  const client = await pool.connect();
  try {
    await client.query("begin");

    let event = null;
    if (command.idempotencyKey) {
      const existing = await client.query(
        "select * from notification_events where idempotency_key = $1",
        [command.idempotencyKey]
      );
      if (existing.rowCount > 0) {
        event = existing.rows[0];
        const deliveries = await loadDeliveriesForEvent(client, event.id);
        await client.query("commit");
        return { ok: true, created: false, idempotent: true, event: toEvent(event), deliveries };
      }
    }

    const eventId = randomUUID();
    const insertedEvent = await client.query(
      `insert into notification_events (
        id,
        type,
        source_service,
        actor_user_id,
        actor_email,
        subject_type,
        subject_id,
        payload,
        idempotency_key
      ) values ($1, $2, $3, $4, $5, $6, $7, $8::jsonb, $9)
      returning *`,
      [
        eventId,
        command.type,
        command.sourceService,
        command.actor.userId,
        command.actor.email,
        command.subject.type,
        command.subject.id,
        JSON.stringify(command.payload),
        command.idempotencyKey,
      ]
    );
    event = insertedEvent.rows[0];

    const deliveries = [];
    for (const delivery of command.deliveries) {
      const insertedDelivery = await client.query(
        `insert into notification_deliveries (
          id,
          event_id,
          recipient_user_id,
          recipient_email,
          surface,
          title,
          body,
          meta,
          status,
          actionable,
          action_type,
          action_url,
          entity_type,
          entity_id
        ) values ($1, $2, $3, $4, $5, $6, $7, $8::jsonb, $9, $10, $11, $12, $13, $14)
        returning *`,
        [
          randomUUID(),
          event.id,
          delivery.recipientUserId,
          delivery.recipientEmail,
          delivery.surface,
          delivery.title,
          delivery.body,
          JSON.stringify(delivery.meta),
          delivery.status,
          delivery.actionable,
          delivery.actionType,
          delivery.actionUrl,
          delivery.entityType,
          delivery.entityId,
        ]
      );
      deliveries.push(toDelivery({ ...insertedDelivery.rows[0], event_type: event.type, source_service: event.source_service, event_payload: event.payload }));
    }

    await client.query("commit");
    for (const delivery of deliveries) {
      broadcastDeliveryUpdate("notification.delivery.created", delivery);
    }
    return { ok: true, created: true, event: toEvent(event), deliveries };
  } catch (error) {
    await client.query("rollback");
    throw error;
  } finally {
    client.release();
  }
}

async function listDeliveries({ subject, surface, limit, unreadOnly }) {
  const values = [surface, subject.userId, subject.email, limit];
  const unreadSql = unreadOnly ? "and d.status = 'unread'" : "";
  const result = await pool.query(
    `select
      d.*,
      e.type as event_type,
      e.source_service,
      e.payload as event_payload
    from notification_deliveries d
    join notification_events e on e.id = d.event_id
    where d.surface = $1
      and (
        ($2::text is not null and d.recipient_user_id = $2)
        or ($3::text is not null and lower(d.recipient_email) = lower($3))
      )
      ${unreadSql}
    order by d.created_at desc
    limit $4`,
    values
  );
  return result.rows.map(toDelivery);
}

async function markDeliveryRead({ subject, deliveryId }) {
  const result = await pool.query(
    `update notification_deliveries
    set status = 'read', read_at = coalesce(read_at, now()), updated_at = now()
    where id = $1
      and status = 'unread'
      and (
        ($2::text is not null and recipient_user_id = $2)
        or ($3::text is not null and lower(recipient_email) = lower($3))
      )
    returning *`,
    [deliveryId, subject.userId, subject.email]
  );
  if (result.rowCount > 0) {
    const delivery = await loadDelivery(result.rows[0].id);
    return delivery;
  }

  const existing = await loadDeliveryForSubject({ subject, deliveryId });
  return existing?.status === "read" ? existing : null;
}

async function markAllDeliveriesRead({ subject, surface }) {
  const result = await pool.query(
    `update notification_deliveries
    set status = 'read', read_at = coalesce(read_at, now()), updated_at = now()
    where surface = $1
      and status = 'unread'
      and (
        ($2::text is not null and recipient_user_id = $2)
        or ($3::text is not null and lower(recipient_email) = lower($3))
      )`,
    [surface, subject.userId, subject.email]
  );
  return { count: result.rowCount };
}

async function loadDeliveriesForEvent(client, eventId) {
  const result = await client.query(
    `select
      d.*,
      e.type as event_type,
      e.source_service,
      e.payload as event_payload
    from notification_deliveries d
    join notification_events e on e.id = d.event_id
    where d.event_id = $1
    order by d.created_at asc`,
    [eventId]
  );
  return result.rows.map(toDelivery);
}

async function loadDelivery(deliveryId) {
  const result = await pool.query(
    `select
      d.*,
      e.type as event_type,
      e.source_service,
      e.payload as event_payload
    from notification_deliveries d
    join notification_events e on e.id = d.event_id
    where d.id = $1`,
    [deliveryId]
  );
  return result.rows[0] ? toDelivery(result.rows[0]) : null;
}

async function loadDeliveryForSubject({ subject, deliveryId }) {
  const result = await pool.query(
    `select
      d.*,
      e.type as event_type,
      e.source_service,
      e.payload as event_payload
    from notification_deliveries d
    join notification_events e on e.id = d.event_id
    where d.id = $1
      and (
        ($2::text is not null and d.recipient_user_id = $2)
        or ($3::text is not null and lower(d.recipient_email) = lower($3))
      )`,
    [deliveryId, subject.userId, subject.email]
  );
  return result.rows[0] ? toDelivery(result.rows[0]) : null;
}

async function migrate() {
  await pool.query(`
    create table if not exists notification_events (
      id uuid primary key,
      type text not null,
      source_service text not null,
      actor_user_id text,
      actor_email text,
      subject_type text,
      subject_id text,
      payload jsonb not null default '{}'::jsonb,
      idempotency_key text unique,
      created_at timestamptz not null default now()
    )
  `);

  await pool.query(`
    create table if not exists notification_deliveries (
      id uuid primary key,
      event_id uuid not null references notification_events(id) on delete cascade,
      recipient_user_id text,
      recipient_email text,
      surface text not null,
      title text not null,
      body text not null default '',
      meta jsonb not null default '{}'::jsonb,
      status text not null default 'unread',
      actionable boolean not null default false,
      action_type text,
      action_url text,
      entity_type text,
      entity_id text,
      created_at timestamptz not null default now(),
      updated_at timestamptz not null default now(),
      read_at timestamptz,
      constraint notification_deliveries_recipient_check
        check (recipient_user_id is not null or recipient_email is not null),
      constraint notification_deliveries_status_check
        check (status in ('unread', 'read', 'archived'))
    )
  `);

  await pool.query("create index if not exists notification_deliveries_recipient_user_idx on notification_deliveries(recipient_user_id, surface, status, created_at desc)");
  await pool.query("create index if not exists notification_deliveries_recipient_email_idx on notification_deliveries(lower(recipient_email), surface, status, created_at desc)");
  await pool.query("create index if not exists notification_events_type_idx on notification_events(type, created_at desc)");
}

function sanitizeEventCommand(payload) {
  const type = requireNonEmptyString(payload?.type, "type");
  const sourceService = normalizeKey(payload?.sourceService || payload?.source_service || "platform");
  const deliveries = Array.isArray(payload?.deliveries) ? payload.deliveries.map(sanitizeDeliveryCommand).filter(Boolean) : [];

  if (deliveries.length === 0) {
    throw badRequest("deliveries_required");
  }

  return {
    type,
    sourceService,
    actor: sanitizeActor(payload?.actor),
    subject: sanitizeSubject(payload?.subject),
    payload: isPlainObject(payload?.payload) ? payload.payload : {},
    deliveries,
    idempotencyKey: optionalString(payload?.idempotencyKey || payload?.idempotency_key),
  };
}

function sanitizeDeliveryCommand(payload) {
  const recipientUserId = optionalString(payload?.recipientUserId || payload?.recipient_user_id);
  const recipientEmail = normalizeEmail(payload?.recipientEmail || payload?.recipient_email);

  if (!recipientUserId && !recipientEmail) {
    return null;
  }

  return {
    recipientUserId,
    recipientEmail,
    surface: sanitizeSurface(payload?.surface || "hub"),
    title: requireNonEmptyString(payload?.title, "delivery.title"),
    body: optionalString(payload?.body) || "",
    meta: isPlainObject(payload?.meta) ? payload.meta : {},
    status: sanitizeStatus(payload?.status || "unread"),
    actionable: payload?.actionable === true,
    actionType: optionalString(payload?.actionType || payload?.action_type),
    actionUrl: optionalString(payload?.actionUrl || payload?.action_url),
    entityType: optionalString(payload?.entityType || payload?.entity_type),
    entityId: optionalString(payload?.entityId || payload?.entity_id),
  };
}

function sanitizeActor(payload) {
  const actor = isPlainObject(payload) ? payload : {};
  return {
    userId: optionalString(actor.userId || actor.user_id),
    email: normalizeEmail(actor.email),
  };
}

function sanitizeSubject(payload) {
  const subject = isPlainObject(payload) ? payload : {};
  return {
    type: optionalString(subject.type),
    id: optionalString(subject.id),
  };
}

function getRequestSubject(req) {
  const userId = optionalString(req.headers["x-nodedc-user-id"] || req.query.userId);
  const email = normalizeEmail(req.headers["x-nodedc-user-email"] || req.query.email);
  if (!userId && !email) {
    throw badRequest("notification_subject_required");
  }
  return { userId, email };
}

function toEvent(row) {
  return {
    id: row.id,
    type: row.type,
    sourceService: row.source_service,
    actorUserId: row.actor_user_id,
    actorEmail: row.actor_email,
    subjectType: row.subject_type,
    subjectId: row.subject_id,
    payload: row.payload || {},
    idempotencyKey: row.idempotency_key,
    createdAt: toIso(row.created_at),
  };
}

function toDelivery(row) {
  return {
    id: row.id,
    eventId: row.event_id,
    eventType: row.event_type,
    sourceService: row.source_service,
    recipientUserId: row.recipient_user_id,
    recipientEmail: row.recipient_email,
    surface: row.surface,
    title: row.title,
    body: row.body,
    meta: row.meta || {},
    status: row.status,
    unread: row.status === "unread",
    actionable: row.actionable,
    actionType: row.action_type,
    actionUrl: row.action_url,
    entityType: row.entity_type,
    entityId: row.entity_id,
    eventPayload: row.event_payload || {},
    createdAt: toIso(row.created_at),
    updatedAt: toIso(row.updated_at),
    readAt: toIso(row.read_at),
  };
}

function broadcastDeliveryUpdate(eventName, delivery) {
  for (const client of streamClients) {
    if (!deliveryMatchesClient(delivery, client)) continue;
    writeSse(client.res, eventName, { delivery });
  }
}

function broadcastSubjectUpdate(eventName, payload) {
  for (const client of streamClients) {
    if (client.surface !== payload.surface) continue;
    if (!subjectsMatch(client.subject, payload.subject)) continue;
    writeSse(client.res, eventName, payload);
  }
}

function deliveryMatchesClient(delivery, client) {
  if (delivery.surface !== client.surface) return false;
  return subjectsMatch(client.subject, {
    userId: delivery.recipientUserId,
    email: delivery.recipientEmail,
  });
}

function subjectsMatch(left, right) {
  return Boolean(
    (left.userId && right.userId && left.userId === right.userId) ||
    (left.email && right.email && left.email.toLowerCase() === right.email.toLowerCase())
  );
}

function writeSse(res, event, data) {
  res.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`);
}

function requireInternalApi(req, res, next) {
  if (!config.internalAccessToken) {
    res.status(503).json({ ok: false, error: "notification_core_internal_token_not_configured" });
    return;
  }

  const authorization = typeof req.headers.authorization === "string" ? req.headers.authorization : "";
  const bearerToken = authorization.match(/^Bearer\s+(.+)$/i)?.[1] || "";
  const headerToken = typeof req.headers["x-nodedc-internal-token"] === "string" ? req.headers["x-nodedc-internal-token"] : "";
  const requestToken = bearerToken || headerToken;

  if (!safeTokenEquals(requestToken, config.internalAccessToken)) {
    res.status(401).json({ ok: false, error: "notification_core_unauthorized" });
    return;
  }

  next();
}

function safeTokenEquals(actual, expected) {
  if (!actual || !expected) return false;
  const actualBuffer = Buffer.from(actual);
  const expectedBuffer = Buffer.from(expected);
  if (actualBuffer.length !== expectedBuffer.length) return false;
  return timingSafeEqual(actualBuffer, expectedBuffer);
}

function sanitizeSurface(value) {
  const surface = normalizeKey(value || "hub");
  if (!["hub", "engine", "operational-core"].includes(surface)) {
    throw badRequest("unsupported_notification_surface");
  }
  return surface;
}

function sanitizeStatus(value) {
  const status = normalizeKey(value || "unread");
  if (!["unread", "read", "archived"].includes(status)) {
    throw badRequest("unsupported_notification_status");
  }
  return status;
}

function sanitizeLimit(value) {
  const limit = Number(value || 100);
  if (!Number.isFinite(limit)) return 100;
  return Math.min(Math.max(Math.trunc(limit), 1), 200);
}

function sanitizeUuid(value, name) {
  const text = optionalString(value);
  if (!/^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i.test(text || "")) {
    throw badRequest(`${name}_invalid`);
  }
  return text;
}

function requireNonEmptyString(value, name) {
  const text = optionalString(value);
  if (!text) throw badRequest(`${name}_required`);
  return text;
}

function optionalString(value) {
  if (typeof value !== "string") return null;
  const text = value.trim();
  return text ? text : null;
}

function normalizeEmail(value) {
  const text = optionalString(value);
  return text ? text.toLowerCase() : null;
}

function normalizeKey(value) {
  return String(value || "").trim().toLowerCase().replace(/_/g, "-");
}

function isPlainObject(value) {
  return Boolean(value && typeof value === "object" && !Array.isArray(value));
}

function toIso(value) {
  if (!value) return null;
  return value instanceof Date ? value.toISOString() : new Date(value).toISOString();
}

function badRequest(message) {
  const error = new Error(message);
  error.status = 400;
  return error;
}

function asyncRoute(handler) {
  return (req, res, next) => {
    Promise.resolve(handler(req, res, next)).catch(next);
  };
}

function readConfig() {
  const databaseUrl =
    process.env.DATABASE_URL ||
    process.env.NOTIFICATION_DATABASE_URL ||
    "postgres://nodedc_notifications:nodedc_notifications@localhost:5432/nodedc_notifications";

  return {
    port: Number(process.env.PORT || process.env.NOTIFICATION_CORE_PORT || "5185"),
    databaseUrl,
    databasePoolSize: Number(process.env.NOTIFICATION_DATABASE_POOL_SIZE || "10"),
    internalAccessToken:
      process.env.NODEDC_INTERNAL_ACCESS_TOKEN ||
      process.env.NODEDC_PLATFORM_SERVICE_TOKEN ||
      "",
  };
}

async function shutdown() {
  try {
    httpServer.close();
    await pool.end();
  } finally {
    process.exit(0);
  }
}