import { eq, and, desc, like, lt, sql, gte, lte, inArray, ne } from 'drizzle-orm';
import type { PgColumn } from 'drizzle-orm/pg-core';

import { db } from '@/lib/db';
import { invoices, invoiceLineItems, invoiceExpenses, expenses } from '@/lib/db/schema/financial';
import { systemSettings } from '@/lib/db/schema/system';
import { clients, clientAddresses } from '@/lib/db/schema/clients';
import { companies, companyAddresses } from '@/lib/db/schema/companies';
import { buildListQuery } from '@/lib/db/query-builder';
import { createAuditLog, toAuditJson, type AuditMeta } from '@/lib/audit';
import { diffEntity } from '@/lib/entity-diff';
import { withTransaction } from '@/lib/db/utils';
import { CodedError, NotFoundError, ConflictError, ValidationError } from '@/lib/errors';
import { getCountryName } from '@/lib/i18n/countries';
import { getSubdivisionName } from '@/lib/i18n/subdivisions';
import { emitToRoom } from '@/lib/socket/server';
import { logger } from '@/lib/logger';
import { getQueue } from '@/lib/queue';
import type {
  CreateInvoiceInput,
  UpdateInvoiceInput,
  RecordPaymentInput,
  ListInvoicesInput,
} from '@/lib/validators/invoices';

// ─── Auto-numbering (BR-041) ───────────────────────────────────────────────

async function generateInvoiceNumber(portId: string, tx: typeof db): Promise<string> {
  const lockKey = `invoice_${portId}`;
  await tx.execute(sql`SELECT pg_advisory_xact_lock(hashtext(${lockKey}))`);

  const now = new Date();
  const prefix = `INV-${now.getFullYear()}${String(now.getMonth() + 1).padStart(2, '0')}`;

  const [existing] = await tx
    .select({ invoiceNumber: invoices.invoiceNumber })
    .from(invoices)
    .where(and(eq(invoices.portId, portId), like(invoices.invoiceNumber, `${prefix}-%`)))
    .orderBy(desc(invoices.invoiceNumber))
    .limit(1);

  let seq = 1;
  if (existing) {
    const parts = existing.invoiceNumber.split('-');
    seq = parseInt(parts[parts.length - 1] ?? '0', 10) + 1;
  }
  return `${prefix}-${String(seq).padStart(3, '0')}`;
}

// ─── Resolve billing entity (polymorphic client | company) ────────────────

/**
 * Look up the billing entity referenced on invoice create and derive the
 * display name + fallback billing email/address. Scoped to the tenant (portId);
 * throws ValidationError on missing / cross-tenant lookups.
 *
 * Runs inside the caller's transaction so the lookup is consistent with the
 * rest of the create operation.
 */
async function resolveBillingEntity(
  tx: typeof db,
  portId: string,
  entity: { type: 'client' | 'company'; id: string },
): Promise<{
  clientName: string;
  billingEmail: string | null;
  billingAddress: string | null;
}> {
  if (entity.type === 'client') {
    const client = await tx.query.clients.findFirst({
      where: and(eq(clients.id, entity.id), eq(clients.portId, portId)),
      with: {
        contacts: true,
      },
    });
    if (!client) throw new ValidationError('billing entity (client) not found');

    // Prefer primary email contact, fall back to any email contact
    const emailContact =
      client.contacts?.find((c) => c.channel === 'email' && c.isPrimary) ??
      client.contacts?.find((c) => c.channel === 'email');

    const addressRow = await tx.query.clientAddresses.findFirst({
      where: and(eq(clientAddresses.clientId, client.id), eq(clientAddresses.isPrimary, true)),
    });
    const billingAddress = addressRow
      ? [
          addressRow.streetAddress,
          addressRow.city,
          addressRow.subdivisionIso ? getSubdivisionName(addressRow.subdivisionIso) : null,
          addressRow.postalCode,
          addressRow.countryIso ? getCountryName(addressRow.countryIso, 'en') : null,
        ]
          .filter(Boolean)
          .join(', ')
      : null;

    return {
      clientName: client.fullName,
      billingEmail: emailContact?.value ?? null,
      billingAddress: billingAddress || null,
    };
  }

  const company = await tx.query.companies.findFirst({
    where: and(eq(companies.id, entity.id), eq(companies.portId, portId)),
  });
  if (!company) throw new ValidationError('billing entity (company) not found');

  const addressRow = await tx.query.companyAddresses.findFirst({
    where: and(eq(companyAddresses.companyId, company.id), eq(companyAddresses.isPrimary, true)),
  });
  const billingAddress = addressRow
    ? [
        addressRow.streetAddress,
        addressRow.city,
        addressRow.subdivisionIso ? getSubdivisionName(addressRow.subdivisionIso) : null,
        addressRow.postalCode,
        addressRow.countryIso ? getCountryName(addressRow.countryIso, 'en') : null,
      ]
        .filter(Boolean)
        .join(', ')
    : null;

  return {
    clientName: company.name,
    billingEmail: company.billingEmail ?? null,
    billingAddress: billingAddress || null,
  };
}

/**
 * Verify every supplied expense ID belongs to the caller's port. Without
 * this gate, a caller could link foreign-port expenses into their own
 * draft invoice and read those expenses back via getInvoiceById's
 * `linkedExpenses` join - a cross-tenant data leak.
 */
async function assertExpensesInPort(
  tx: typeof db,
  portId: string,
  expenseIds: string[],
): Promise<void> {
  if (expenseIds.length === 0) return;
  const rows = await tx
    .select({ id: expenses.id })
    .from(expenses)
    .where(and(inArray(expenses.id, expenseIds), eq(expenses.portId, portId)));
  if (rows.length !== expenseIds.length) {
    throw new ValidationError('One or more expenses not found in this port');
  }
}

// ─── List ─────────────────────────────────────────────────────────────────

export async function listInvoices(portId: string, query: ListInvoicesInput) {
  const filters = [];

  if (query.status) {
    filters.push(eq(invoices.status, query.status));
  }
  if (query.clientName) {
    filters.push(like(invoices.clientName, `%${query.clientName}%`));
  }
  if (query.dateFrom) {
    filters.push(gte(invoices.dueDate, query.dateFrom));
  }
  if (query.dateTo) {
    filters.push(lte(invoices.dueDate, query.dateTo));
  }
  if (query.billingEntityType) {
    filters.push(eq(invoices.billingEntityType, query.billingEntityType));
  }
  if (query.billingEntityId) {
    filters.push(eq(invoices.billingEntityId, query.billingEntityId));
  }

  return buildListQuery({
    table: invoices,
    portIdColumn: invoices.portId,
    portId,
    idColumn: invoices.id,
    updatedAtColumn: invoices.updatedAt,
    filters,
    page: query.page,
    pageSize: query.limit,
    searchColumns: [invoices.clientName, invoices.invoiceNumber],
    searchTerm: query.search,
    includeArchived: query.includeArchived,
    archivedAtColumn: invoices.archivedAt,
    sort: query.sort
      ? {
          column: invoices[query.sort as keyof typeof invoices] as unknown as PgColumn,
          direction: query.order,
        }
      : undefined,
  });
}

// ─── Get by ID ────────────────────────────────────────────────────────────

export async function getInvoiceById(id: string, portId: string) {
  const invoice = await db.query.invoices.findFirst({
    where: and(eq(invoices.id, id), eq(invoices.portId, portId)),
  });
  if (!invoice) throw new NotFoundError('Invoice');

  const lineItems = await db
    .select()
    .from(invoiceLineItems)
    .where(eq(invoiceLineItems.invoiceId, id))
    .orderBy(invoiceLineItems.sortOrder);

  // Defense-in-depth: even if a join row somehow points at a foreign-tenant
  // expense, the WHERE clause filters by expenses.portId so cross-tenant data
  // can't leak through this read.
  const linkedExpenses = await db
    .select({ expense: expenses })
    .from(invoiceExpenses)
    .innerJoin(expenses, eq(expenses.id, invoiceExpenses.expenseId))
    .where(and(eq(invoiceExpenses.invoiceId, id), eq(expenses.portId, portId)));

  return {
    ...invoice,
    lineItems,
    linkedExpenses: linkedExpenses.map((r) => r.expense),
  };
}

// ─── Create (BR-041, BR-042, BR-045) ─────────────────────────────────────

export async function createInvoice(portId: string, data: CreateInvoiceInput, meta: AuditMeta) {
  const invoice = await withTransaction(async (tx) => {
    // Resolve the polymorphic billing entity (client | company). Throws
    // ValidationError if the entity is missing or belongs to another tenant.
    const entitySnapshot = await resolveBillingEntity(tx, portId, data.billingEntity);
    // clientName is always entity-derived on create (it's a snapshot).
    // Caller-supplied billingEmail/billingAddress win over entity-derived values.
    const effectiveClientName = entitySnapshot.clientName;
    const effectiveBillingEmail = data.billingEmail ?? entitySnapshot.billingEmail;
    const effectiveBillingAddress = data.billingAddress ?? entitySnapshot.billingAddress;

    const invoiceNumber = await generateInvoiceNumber(portId, tx);

    // Calculate subtotal from line items. The `z.coerce.number()` in
    // the schema makes the parsed value a number at runtime - narrow
    // the post-parse shape locally so v4's stricter input typing
    // (unknown for coerced fields) doesn't leak into arithmetic.
    type ParsedLineItem = { quantity: number; unitPrice: number; description: string };
    const lineItemsData = (data.lineItems ?? []) as ParsedLineItem[];
    const subtotal = lineItemsData.reduce((sum, li) => sum + (li.quantity ?? 1) * li.unitPrice, 0);

    // BR-042: net10 discount - read from systemSettings
    let discountPct = 0;
    if (data.paymentTerms === 'net10') {
      const [setting] = await tx
        .select({ value: systemSettings.value })
        .from(systemSettings)
        .where(
          and(eq(systemSettings.key, 'invoice_net10_discount'), eq(systemSettings.portId, portId)),
        )
        .limit(1);

      if (setting) {
        discountPct = Number(setting.value) || 2;
      } else {
        discountPct = 2;
      }
    }

    const discountAmount = (subtotal * discountPct) / 100;
    const feeAmount = 0; // No fee by default
    const feePct = 0;
    const total = subtotal - discountAmount + feeAmount;

    // BR-045: Verify expenses aren't already linked to a non-draft invoice.
    // Tenancy guard precedes BR-045 so a foreign-port expense fails with
    // ValidationError before any further checks (or any join-side leak).
    const expenseIds = data.expenseIds ?? [];
    await assertExpensesInPort(tx, portId, expenseIds);
    if (expenseIds.length > 0) {
      const alreadyLinked = await tx
        .select({ expenseId: invoiceExpenses.expenseId })
        .from(invoiceExpenses)
        .innerJoin(invoices, eq(invoices.id, invoiceExpenses.invoiceId))
        .where(
          and(inArray(invoiceExpenses.expenseId, expenseIds), sql`${invoices.status} != 'draft'`),
        )
        .limit(1);

      if (alreadyLinked.length > 0) {
        throw new ConflictError('One or more expenses are already linked to a non-draft invoice');
      }
    }

    // Sanity-check the optional interest link: must belong to the same port.
    // Foreign-port ids fail with ValidationError before the insert.
    if (data.interestId) {
      const { interests } = await import('@/lib/db/schema/interests');
      const [interestRow] = await tx
        .select({ portId: interests.portId })
        .from(interests)
        .where(eq(interests.id, data.interestId))
        .limit(1);
      if (!interestRow || interestRow.portId !== portId) {
        throw new ValidationError('interestId not found in this port');
      }
    }

    const [newInvoice] = await tx
      .insert(invoices)
      .values({
        portId,
        invoiceNumber,
        billingEntityType: data.billingEntity.type,
        billingEntityId: data.billingEntity.id,
        clientName: effectiveClientName,
        billingEmail: effectiveBillingEmail,
        billingAddress: effectiveBillingAddress,
        dueDate: data.dueDate,
        paymentTerms: data.paymentTerms ?? 'net30',
        currency: data.currency ?? 'USD',
        subtotal: String(subtotal),
        discountPct: String(discountPct),
        discountAmount: String(discountAmount),
        feePct: String(feePct),
        feeAmount: String(feeAmount),
        total: String(total),
        status: 'draft',
        paymentStatus: 'unpaid',
        interestId: data.interestId ?? null,
        kind: data.kind ?? 'general',
        notes: data.notes ?? null,
        createdBy: meta.userId,
      })
      .returning();

    if (!newInvoice)
      throw new CodedError('INSERT_RETURNING_EMPTY', {
        internalMessage: 'Invoice insert returned no row',
      });

    // Insert line items
    if (lineItemsData.length > 0) {
      await tx.insert(invoiceLineItems).values(
        lineItemsData.map((li, idx) => ({
          invoiceId: newInvoice.id,
          description: li.description,
          quantity: String(li.quantity ?? 1),
          unitPrice: String(li.unitPrice),
          total: String((li.quantity ?? 1) * li.unitPrice),
          sortOrder: idx,
        })),
      );
    }

    // Link expenses
    if (expenseIds.length > 0) {
      await tx.insert(invoiceExpenses).values(
        expenseIds.map((expenseId) => ({
          invoiceId: newInvoice.id,
          expenseId,
        })),
      );
    }

    return newInvoice;
  });

  void createAuditLog({
    userId: meta.userId,
    portId,
    action: 'create',
    entityType: 'invoice',
    entityId: invoice.id,
    newValue: toAuditJson(invoice),
    ipAddress: meta.ipAddress,
    userAgent: meta.userAgent,
  });

  emitToRoom(`port:${portId}`, 'invoice:created', {
    invoiceId: invoice.id,
    invoiceNumber: invoice.invoiceNumber,
    total: Number(invoice.total),
    clientName: invoice.clientName,
  });

  return invoice;
}

// ─── Update (draft only) ──────────────────────────────────────────────────

export async function updateInvoice(
  id: string,
  portId: string,
  data: UpdateInvoiceInput,
  meta: AuditMeta,
) {
  const existing = await getInvoiceById(id, portId);
  if (existing.status !== 'draft') {
    throw new ConflictError('Only draft invoices can be updated');
  }

  const updated = await withTransaction(async (tx) => {
    const updateData: Record<string, unknown> = { updatedAt: new Date() };

    if (data.clientName !== undefined) updateData.clientName = data.clientName;
    if (data.billingEmail !== undefined) updateData.billingEmail = data.billingEmail;
    if (data.billingAddress !== undefined) updateData.billingAddress = data.billingAddress;
    if (data.dueDate !== undefined) updateData.dueDate = data.dueDate;
    if (data.paymentTerms !== undefined) updateData.paymentTerms = data.paymentTerms;
    if (data.currency !== undefined) updateData.currency = data.currency;
    if (data.notes !== undefined) updateData.notes = data.notes;
    if (data.interestId !== undefined) {
      if (data.interestId !== null) {
        const { interests } = await import('@/lib/db/schema/interests');
        const [interestRow] = await tx
          .select({ portId: interests.portId })
          .from(interests)
          .where(eq(interests.id, data.interestId))
          .limit(1);
        if (!interestRow || interestRow.portId !== portId) {
          throw new ValidationError('interestId not found in this port');
        }
      }
      updateData.interestId = data.interestId;
    }
    if (data.kind !== undefined) updateData.kind = data.kind;

    // Recalculate totals if line items changed (see createInvoice for
    // the ParsedLineItem narrowing rationale - same coerced-number
    // story applies on the update path).
    if (data.lineItems !== undefined) {
      type ParsedLineItem = { quantity: number; unitPrice: number; description: string };
      const lineItemsData = data.lineItems as ParsedLineItem[];
      const subtotal = lineItemsData.reduce(
        (sum, li) => sum + (li.quantity ?? 1) * li.unitPrice,
        0,
      );

      const paymentTerms = data.paymentTerms ?? existing.paymentTerms;
      let discountPct = 0;
      if (paymentTerms === 'net10') {
        const [setting] = await tx
          .select({ value: systemSettings.value })
          .from(systemSettings)
          .where(
            and(
              eq(systemSettings.key, 'invoice_net10_discount'),
              eq(systemSettings.portId, portId),
            ),
          )
          .limit(1);
        discountPct = setting ? Number(setting.value) || 2 : 2;
      }

      const discountAmount = (subtotal * discountPct) / 100;
      const feeAmount = Number(existing.feeAmount) || 0;
      const feePct = Number(existing.feePct) || 0;
      const total = subtotal - discountAmount + feeAmount;

      updateData.subtotal = String(subtotal);
      updateData.discountPct = String(discountPct);
      updateData.discountAmount = String(discountAmount);
      updateData.feePct = String(feePct);
      updateData.feeAmount = String(feeAmount);
      updateData.total = String(total);

      // Replace line items
      await tx.delete(invoiceLineItems).where(eq(invoiceLineItems.invoiceId, id));
      if (lineItemsData.length > 0) {
        await tx.insert(invoiceLineItems).values(
          lineItemsData.map((li, idx) => ({
            invoiceId: id,
            description: li.description,
            quantity: String(li.quantity ?? 1),
            unitPrice: String(li.unitPrice),
            total: String((li.quantity ?? 1) * li.unitPrice),
            sortOrder: idx,
          })),
        );
      }
    }

    // Replace expense links if provided
    if (data.expenseIds !== undefined) {
      // Tenancy gate first - reject foreign-port expense IDs before
      // running BR-045 or doing any writes.
      await assertExpensesInPort(tx, portId, data.expenseIds);
      // BR-045
      if (data.expenseIds.length > 0) {
        const alreadyLinked = await tx
          .select({ expenseId: invoiceExpenses.expenseId })
          .from(invoiceExpenses)
          .innerJoin(invoices, eq(invoices.id, invoiceExpenses.invoiceId))
          .where(
            and(
              inArray(invoiceExpenses.expenseId, data.expenseIds),
              sql`${invoices.status} != 'draft'`,
              ne(invoices.id, id),
            ),
          )
          .limit(1);

        if (alreadyLinked.length > 0) {
          throw new ConflictError('One or more expenses are already linked to a non-draft invoice');
        }
      }

      await tx.delete(invoiceExpenses).where(eq(invoiceExpenses.invoiceId, id));
      if (data.expenseIds.length > 0) {
        await tx
          .insert(invoiceExpenses)
          .values(data.expenseIds.map((expenseId) => ({ invoiceId: id, expenseId })));
      }
    }

    const [result] = await tx
      .update(invoices)
      .set(updateData as Record<string, unknown>)
      .where(and(eq(invoices.id, id), eq(invoices.portId, portId)))
      .returning();

    if (!result) throw new NotFoundError('Invoice');
    return result;
  });

  const { diff } = diffEntity(toAuditJson(existing), toAuditJson(updated));

  void createAuditLog({
    userId: meta.userId,
    portId,
    action: 'update',
    entityType: 'invoice',
    entityId: id,
    oldValue: toAuditJson(existing),
    newValue: toAuditJson(updated),
    metadata: { diff },
    ipAddress: meta.ipAddress,
    userAgent: meta.userAgent,
  });

  emitToRoom(`port:${portId}`, 'invoice:updated', {
    invoiceId: id,
    changedFields: Object.keys(diff),
  });

  return updated;
}

// ─── Delete (draft only) ──────────────────────────────────────────────────

export async function deleteInvoice(id: string, portId: string, meta: AuditMeta) {
  const existing = await getInvoiceById(id, portId);
  if (existing.status !== 'draft') {
    throw new ConflictError('Only draft invoices can be deleted');
  }

  await withTransaction(async (tx) => {
    await tx.delete(invoiceExpenses).where(eq(invoiceExpenses.invoiceId, id));
    await tx.delete(invoiceLineItems).where(eq(invoiceLineItems.invoiceId, id));
    await tx.delete(invoices).where(and(eq(invoices.id, id), eq(invoices.portId, portId)));
  });

  void createAuditLog({
    userId: meta.userId,
    portId,
    action: 'delete',
    entityType: 'invoice',
    entityId: id,
    oldValue: toAuditJson(existing),
    ipAddress: meta.ipAddress,
    userAgent: meta.userAgent,
  });

  emitToRoom(`port:${portId}`, 'invoice:updated', {
    invoiceId: id,
    changedFields: ['status'],
  });
}

// ─── Send invoice ─────────────────────────────────────────────────────────

export async function sendInvoice(id: string, portId: string, meta: AuditMeta) {
  const invoice = await getInvoiceById(id, portId);

  // Invoice PDF generation has been removed (the CRM no longer renders
  // client-facing PDFs from scratch - see the PDF stack overhaul spec).
  // The "send" event still fires so the queue + audit + socket flow
  // remains intact; downstream consumers can decide whether to render
  // an external document, link to the in-app view, or wait for the
  // admin-uploaded AcroForm-fill feature to ship.
  // Stable jobId for natural dedup: a double-click on the Send button
  // or a webhook retry on the upstream caller can fire this twice. BullMQ
  // rejects a duplicate `jobId` while the original is still queued or
  // active, so we get at-most-once email per invoice-send action.
  // concurrency-auditor C-2.
  await getQueue('email').add(
    'send-invoice',
    { invoiceId: id, portId },
    { jobId: `send-invoice:${id}` },
  );

  // Update status to 'sent'
  const [updated] = await db
    .update(invoices)
    .set({ status: 'sent', updatedAt: new Date() })
    .where(and(eq(invoices.id, id), eq(invoices.portId, portId)))
    .returning();

  void createAuditLog({
    userId: meta.userId,
    portId,
    action: 'update',
    entityType: 'invoice',
    entityId: id,
    oldValue: { status: invoice.status },
    newValue: { status: 'sent' },
    metadata: { action: 'invoice_sent' },
    ipAddress: meta.ipAddress,
    userAgent: meta.userAgent,
  });

  emitToRoom(`port:${portId}`, 'invoice:sent', {
    invoiceId: id,
    invoiceNumber: invoice.invoiceNumber,
    recipientEmail: invoice.billingEmail ?? '',
  });

  return updated;
}

// ─── Record payment ───────────────────────────────────────────────────────

export async function recordPayment(
  id: string,
  portId: string,
  data: RecordPaymentInput,
  meta: AuditMeta,
) {
  const existing = await getInvoiceById(id, portId);

  const [updated] = await db
    .update(invoices)
    .set({
      paymentStatus: 'paid',
      paymentDate: data.paymentDate,
      paymentMethod: data.paymentMethod ?? null,
      paymentReference: data.paymentReference ?? null,
      status: 'paid',
      updatedAt: new Date(),
    })
    .where(and(eq(invoices.id, id), eq(invoices.portId, portId)))
    .returning();

  if (!updated) throw new NotFoundError('Invoice');

  void createAuditLog({
    userId: meta.userId,
    portId,
    action: 'update',
    entityType: 'invoice',
    entityId: id,
    oldValue: { status: existing.status, paymentStatus: existing.paymentStatus },
    newValue: { status: 'paid', paymentStatus: 'paid', paymentDate: data.paymentDate },
    metadata: { action: 'payment_recorded' },
    ipAddress: meta.ipAddress,
    userAgent: meta.userAgent,
  });

  emitToRoom(`port:${portId}`, 'invoice:paid', {
    invoiceId: id,
    invoiceNumber: existing.invoiceNumber,
    amount: Number(existing.total),
  });

  // Deposit invoices linked to a sales interest auto-advance the pipeline.
  // Only advances forward - no-op if the interest has already moved past
  // deposit_paid (e.g. straight-to-contract flows). NOTE: the v1 sales
  // refactor introduces a separate `payments` table that supersedes invoice
  // tracking for the deposit stage; this block stays wired for legacy
  // invoices but new flows record payments via that pathway instead.
  if (updated.kind === 'deposit' && updated.interestId) {
    const { advanceStageIfBehindGated } = await import('@/lib/services/interests.service');
    void advanceStageIfBehindGated(
      updated.interestId,
      portId,
      'deposit_paid',
      meta,
      `Deposit invoice ${existing.invoiceNumber} paid`,
      'deposit_received',
    );

    // Deposit-paid also fires the berth-rule for `deposit_received` so admins
    // can auto-mark the primary berth as Sold (default rule mode: 'auto').
    // Dynamic import keeps the invoices ↔ berth-rules graph one-way and
    // mirrors the existing advanceStageIfBehind dispatch above.
    const { evaluateRule } = await import('@/lib/services/berth-rules-engine');
    void evaluateRule('deposit_received', updated.interestId, portId, meta);
  }

  return updated;
}

// ─── Detect overdue (BR-044) ──────────────────────────────────────────────

export async function detectOverdue(portId: string) {
  const today = new Date().toISOString().split('T')[0]!;

  const overdueInvoices = await db
    .select({ id: invoices.id, invoiceNumber: invoices.invoiceNumber, dueDate: invoices.dueDate })
    .from(invoices)
    .where(
      and(eq(invoices.portId, portId), eq(invoices.status, 'sent'), lt(invoices.dueDate, today)),
    );

  if (overdueInvoices.length === 0) return;

  for (const inv of overdueInvoices) {
    await db
      .update(invoices)
      .set({ status: 'overdue', updatedAt: new Date() })
      .where(and(eq(invoices.id, inv.id), eq(invoices.portId, portId)));

    const daysPastDue = Math.max(
      1,
      Math.ceil((Date.now() - new Date(inv.dueDate).getTime()) / (1000 * 60 * 60 * 24)),
    );
    emitToRoom(`port:${portId}`, 'invoice:overdue', {
      invoiceId: inv.id,
      invoiceNumber: inv.invoiceNumber,
      daysPastDue,
    });

    // Stable jobId: detectOverdue runs daily; if it fires twice in
    // the same UTC day (e.g. a manual re-trigger after a worker
    // restart) we don't want duplicate overdue emails. Per-day key
    // gives idempotency for the daily fire while letting tomorrow's
    // run re-notify if the invoice still hasn't been paid.
    const dayKey = new Date().toISOString().slice(0, 10);
    await getQueue('notifications').add(
      'invoice-overdue-notify',
      { invoiceId: inv.id, portId },
      { jobId: `invoice-overdue-notify:${inv.id}:${dayKey}` },
    );

    logger.info(
      { invoiceId: inv.id, invoiceNumber: inv.invoiceNumber, portId },
      'Invoice marked overdue',
    );
  }
}