/**
 * DNS registration ledger operations
 * (designs/DISPATCHER_DAEMON_AND_TIMER_EVENTS.md B2/B3).
 *
 * The capability loader records every successful
 * dns_registrar.registerHost here; the run-hook path reads the ledger
 * back to feed a provider's `refresh_registrations` hook, the
 * `public_dns` check resolves every row from off-fleet, and `celilo dns
 * registrations` lists it for the operator.
 *
 * The ledger records WHICH MODULE ASKED FOR WHICH NAME. It stores no
 * address — see the schema comment and design.md D1.
 *
 * Lifecycle: a row dies with its provider by FK cascade, and with its
 * LAST consumer via the `dns_registration_consumers` set (design.md D5).
 */

import type { DnsRegistrarCapability, HookResult } from '@celilo/capabilities';
import { and, eq, sql } from 'drizzle-orm';
import type { DbClient } from '../db/client';
import { dnsRegistrationConsumers, dnsRegistrations } from '../db/schema';

export interface DnsRegistrationRow {
  fqdn: string;
  providerModuleId: string;
  /** The module that introduced the name — the earliest consumer. */
  consumerModuleId: string;
  /** Every module currently depending on the name, introducer first. */
  consumerModuleIds: string[];
  /** Claimed by celilo as the companion of a declared name, not asked for. */
  companion: boolean;
  registeredAt: Date;
  refreshedAt: Date | null;
}

/**
 * Drop registrations whose last consumer module is gone.
 *
 * ponytail: pruned on read rather than by trigger — SQLite fires delete
 * triggers for FK-cascaded deletes only when `recursive_triggers` is on,
 * and reads are the only thing that consumes the ledger. Move it into a
 * trigger if something ever reads these rows without going through here.
 *
 * It CHECKS before deleting, so the overwhelmingly common case (nothing
 * orphaned) stays a pure read. The first version ran the DELETE
 * unconditionally, which took a write lock on every list — including the
 * refresh hook's, and `celilo dns registrations`. A read that quietly writes
 * is a surprise on its own, and on SQLite it is a surprise that serialises
 * against every other writer for no benefit.
 */
function pruneOrphanedRegistrations(db: DbClient): void {
  const orphaned = sql`SELECT 1 FROM dns_registrations WHERE NOT EXISTS (
      SELECT 1 FROM dns_registration_consumers c WHERE c.registration_id = dns_registrations.id
    ) LIMIT 1`;
  if (!db.get(orphaned)) return;

  db.run(
    sql`DELETE FROM dns_registrations WHERE NOT EXISTS (
      SELECT 1 FROM dns_registration_consumers c WHERE c.registration_id = dns_registrations.id
    )`,
  );
}

export function listDnsRegistrations(
  db: DbClient,
  options: { providerModuleId?: string } = {},
): DnsRegistrationRow[] {
  pruneOrphanedRegistrations(db);

  const query = db
    .select({
      id: dnsRegistrations.id,
      fqdn: dnsRegistrations.fqdn,
      providerModuleId: dnsRegistrations.providerModuleId,
      companion: dnsRegistrations.companion,
      registeredAt: dnsRegistrations.registeredAt,
      refreshedAt: dnsRegistrations.refreshedAt,
    })
    .from(dnsRegistrations);
  const rows = options.providerModuleId
    ? query.where(eq(dnsRegistrations.providerModuleId, options.providerModuleId)).all()
    : query.all();

  const consumers = db
    .select({
      registrationId: dnsRegistrationConsumers.registrationId,
      moduleId: dnsRegistrationConsumers.moduleId,
    })
    .from(dnsRegistrationConsumers)
    .orderBy(dnsRegistrationConsumers.id)
    .all();

  const byRegistration = new Map<number, string[]>();
  for (const c of consumers) {
    const list = byRegistration.get(c.registrationId);
    if (list) list.push(c.moduleId);
    else byRegistration.set(c.registrationId, [c.moduleId]);
  }

  return rows.map(({ id, ...row }) => {
    const consumerModuleIds = byRegistration.get(id) ?? [];
    return {
      ...row,
      consumerModuleIds,
      consumerModuleId: consumerModuleIds[0] ?? '',
    };
  });
}

export function recordDnsRegistration(
  db: DbClient,
  registration: {
    providerModuleId: string;
    consumerModuleId: string;
    fqdn: string;
    companion?: boolean;
  },
): void {
  db.insert(dnsRegistrations)
    .values({
      providerModuleId: registration.providerModuleId,
      fqdn: registration.fqdn,
      companion: registration.companion ?? false,
      registeredAt: new Date(),
    })
    .onConflictDoUpdate({
      target: [dnsRegistrations.providerModuleId, dnsRegistrations.fqdn],
      // `companion` is not re-asserted: once a module declares a name
      // outright it stops being something celilo claimed on its behalf.
      set: { registeredAt: new Date() },
    })
    .run();

  const row = db
    .select({ id: dnsRegistrations.id })
    .from(dnsRegistrations)
    .where(
      and(
        eq(dnsRegistrations.providerModuleId, registration.providerModuleId),
        eq(dnsRegistrations.fqdn, registration.fqdn),
      ),
    )
    .get();
  if (!row) return;

  // A re-assert ADDS the asserting module rather than replacing whoever
  // was there — the row must outlive any single one of them (D5).
  db.insert(dnsRegistrationConsumers)
    .values({
      registrationId: row.id,
      moduleId: registration.consumerModuleId,
      firstSeenAt: new Date(),
    })
    .onConflictDoNothing()
    .run();
}

/** Stamp every row of a provider as refreshed (successful refresh hook). */
export function stampDnsRegistrationsRefreshed(db: DbClient, providerModuleId: string): void {
  db.update(dnsRegistrations)
    .set({ refreshedAt: new Date() })
    .where(eq(dnsRegistrations.providerModuleId, providerModuleId))
    .run();
}

/**
 * Wrap a dns_registrar interface so successful registerHost calls are
 * recorded in the ledger. Failures and MissingProviderInputError
 * interview throws pass through untouched — only a confirmed
 * registration earns a row.
 *
 * A provider that ATTEMPTED a companion name (`www.<domain>` ↔
 * `<domain>`) reports it as `outputs.companion_fqdn`, and it gets its own
 * row whether or not the attempt reported success. The row is not a claim
 * that the name is published — it is what puts the name under the
 * `public_dns` check, which is the only thing that can tell. Namecheap
 * returns `ErrCount 0` for a `www` update it silently does not apply, so
 * a companion recorded only on "success" would be a name celilo believes
 * it owns and never looks at again (design.md D3).
 */
export function withDnsRegistrationLedger(
  registrar: DnsRegistrarCapability,
  ctx: { db: DbClient; providerModuleId: string; consumerModuleId: string },
): DnsRegistrarCapability {
  return {
    ...registrar,
    async registerHost(request): Promise<HookResult> {
      const result = await registrar.registerHost(request);
      if (result.success && request.fqdn) {
        recordDnsRegistration(ctx.db, {
          providerModuleId: ctx.providerModuleId,
          consumerModuleId: ctx.consumerModuleId,
          fqdn: request.fqdn,
        });
        const companion = result.outputs?.companion_fqdn;
        if (typeof companion === 'string' && companion.length > 0) {
          recordDnsRegistration(ctx.db, {
            providerModuleId: ctx.providerModuleId,
            consumerModuleId: ctx.consumerModuleId,
            fqdn: companion,
            companion: true,
          });
        }
      }
      return result;
    },
  };
}