/**
 * Port-forward registry — the DB-backed `PortForwardStore` (openspec/changes/unified-management-no-ssh/proposal.md).
 *
 * The shared-core desired-state store for the `firewall` capability. The
 * capability-loader constructs one of these bound to the CONSUMING module and
 * injects it into the firewall provider factory, so `exposeService` becomes a
 * declaration of that consumer's complete port set and the provider's converge
 * renders the whole ruleset from `list(firewallIp)`, applying it atomically
 * (`iptables-restore`). Replaces "read the box back with `iptables -L`" as the
 * source of truth.
 *
 * `registeredBy` is stamped HERE, never accepted from a caller — the same rule
 * `buildTrustedSourceStore` follows, so a forward cannot be attributed to the
 * wrong module (openspec/changes/consumer-removal-cleanup, D2).
 */

import type { PortForward, PortForwardStore, PortForwardTarget } from '@celilo/capabilities';
import { and, eq, isNull, or } from 'drizzle-orm';
import type { DbClient } from '../db/client';
import { portForwards } from '../db/schema';

/** NULL-aware match on the optional ingress IP (SQLite treats NULL as distinct). */
function ingressMatch(ingressIp: string | undefined) {
  return ingressIp ? eq(portForwards.ingressIp, ingressIp) : isNull(portForwards.ingressIp);
}

/**
 * Every row this consumer owns for one target, plus the UNATTRIBUTED rows for
 * the same target.
 *
 * The unattributed half is the migration's other end: rows written before
 * `registered_by` existed carry `''`, and a consumer re-declaring the target
 * they describe is the one moment we can safely say who owns them — one module
 * owns a backend IP. Without this they would render forever with no owner to
 * withdraw them.
 */
function ownedOrUnattributed(firewallIp: string, target: PortForwardTarget, consumer: string) {
  return and(
    eq(portForwards.firewallIp, firewallIp),
    eq(portForwards.internalIp, target.internalIp),
    eq(portForwards.protocol, target.protocol),
    ingressMatch(target.ingressIp),
    or(eq(portForwards.registeredBy, consumer), eq(portForwards.registeredBy, '')),
  );
}

/**
 * Build the store bound to the module that will call through the capability.
 *
 * ponytail: a consumer that stops exposing a target ENTIRELY — its backend IP
 * changes, or it drops a host — leaves rows for the old target, because no call
 * arrives to declare that target's set empty. Bounded: the rows die when the
 * module is removed, and a redeploy onto a new host is the only way to reach
 * it. Closing it needs a sweep against `getModuleSystems`, which is not
 * obviously worth its own failure mode.
 */
export function buildPortForwardStore(db: DbClient, registeredBy: string): PortForwardStore {
  return {
    list(firewallIp: string): PortForward[] {
      return db
        .select()
        .from(portForwards)
        .where(eq(portForwards.firewallIp, firewallIp))
        .all()
        .map((r) => ({
          internalIp: r.internalIp,
          port: r.port,
          protocol: r.protocol,
          ingressIp: r.ingressIp ?? undefined,
          description: r.description,
          registeredBy: r.registeredBy,
        }));
    },

    replace(firewallIp: string, target: PortForwardTarget, ports: number[]): void {
      // Delete-then-insert over the consumer's WHOLE set for this target, not
      // per port: that is what makes a redeploy exposing a shorter port list
      // withdraw the ports it left out (celilo#855). Scoped to this consumer,
      // so another module's forward for the same target is untouched.
      db.delete(portForwards)
        .where(ownedOrUnattributed(firewallIp, target, registeredBy))
        .run();

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

      db.insert(portForwards)
        .values(
          ports.map((port) => ({
            firewallIp,
            internalIp: target.internalIp,
            port,
            protocol: target.protocol,
            ingressIp: target.ingressIp ?? null,
            description: target.description,
            registeredBy,
          })),
        )
        .run();
    },
  };
}