feat(health-checks): enhance health check functionality and add new routes

- Introduced origin health check routes and integrated them into the application.
- Updated health check configuration to include success recovery thresholds.
- Expanded error handling with new error codes for health check failures.
- Added new service routes for managing health checks, including creation and listing.
- Improved health check service logic to track consecutive successes and failures.

This commit enhances the health check capabilities, providing better monitoring and management of service health.

packages/db/src/repos.ts

@@ -11,17 +11,19 @@ import type {
   IpHealthState,
   IpHealthStatus,
   LbMode,
+  OriginHealthCheck,
   Service,
   ServiceBinding,
   ServiceBindingView,
   ServiceGroup,
+  ServiceNode,
   Subdomain,
   SyncJob,
 } from "@cfdm/shared";
 import { dnsRecordNamesMatch, isIpLiteral } from "@cfdm/shared";
 import { and, asc, count, eq, isNull, like, notInArray, or, sql } from "drizzle-orm";
 import type { Db } from "./client.js";
-import { NotFoundError } from "./errors.js";
+import { ConflictError, NotFoundError } from "./errors.js";
 import {
   certificates,
   dnsRecords,
@@ -30,7 +32,10 @@ import {
   domainTags,
   domains,
   groups,
+  healthChecks,
   ipHealthStatus,
+  nodes,
+  bindingNodes,
   notificationLog,
   serviceBindingIps,
   serviceBindingRecords,
@@ -911,6 +916,17 @@ export function deleteServiceGroup(db: Db, id: number): void {
 
 // --- Service IPs ---
 
+function insertServiceIpIfMissing(db: Db, serviceId: number, ip: string): void {
+  const existing = db
+    .select({ ip: serviceIps.ip })
+    .from(serviceIps)
+    .where(and(eq(serviceIps.service_id, serviceId), eq(serviceIps.ip, ip)))
+    .get();
+  if (!existing) {
+    db.insert(serviceIps).values({ service_id: serviceId, ip }).run();
+  }
+}
+
 export function listServiceIps(db: Db, serviceId: number): string[] {
   return db
     .select({ ip: serviceIps.ip })
@@ -928,7 +944,363 @@ export function replaceServiceIps(
   db.delete(serviceIps).where(eq(serviceIps.service_id, serviceId)).run();
   for (const ip of ips) {
     db.insert(serviceIps).values({ service_id: serviceId, ip }).run();
+    ensureNode(db, serviceId, ip);
   }
+  const keep = new Set(ips);
+  for (const node of listNodes(db, serviceId)) {
+    if (keep.has(node.address)) continue;
+    const bound = db
+      .select({ node_id: bindingNodes.node_id })
+      .from(bindingNodes)
+      .where(eq(bindingNodes.node_id, node.id))
+      .get();
+    if (!bound) deleteNode(db, node.id);
+  }
+}
+
+function mapNode(row: typeof nodes.$inferSelect): ServiceNode {
+  return {
+    id: row.id,
+    service_id: row.service_id,
+    address: row.address,
+    protocol: row.protocol,
+    port: row.port,
+    enabled: Boolean(row.enabled),
+    priority: row.priority,
+    weight: row.weight,
+    health_status: row.health_status as ServiceNode["health_status"],
+    health_check_id: row.health_check_id,
+    consecutive_failures: row.consecutive_failures,
+    consecutive_successes: row.consecutive_successes,
+    last_check_at: row.last_check_at,
+    last_failure_reason: row.last_failure_reason,
+    created_at: row.created_at,
+    updated_at: row.updated_at,
+  };
+}
+
+export function listNodes(db: Db, serviceId: number): ServiceNode[] {
+  return db
+    .select()
+    .from(nodes)
+    .where(eq(nodes.service_id, serviceId))
+    .all()
+    .map(mapNode);
+}
+
+export function getNode(db: Db, id: number): ServiceNode {
+  const row = db.select().from(nodes).where(eq(nodes.id, id)).get();
+  if (!row) throw new NotFoundError(`node ${id}`);
+  return mapNode(row);
+}
+
+export function findNodeByAddress(
+  db: Db,
+  serviceId: number,
+  address: string,
+): ServiceNode | null {
+  const row = db
+    .select()
+    .from(nodes)
+    .where(and(eq(nodes.service_id, serviceId), eq(nodes.address, address)))
+    .get();
+  return row ? mapNode(row) : null;
+}
+
+export function findNodeByIp(db: Db, address: string): ServiceNode | null {
+  const row = db.select().from(nodes).where(eq(nodes.address, address)).get();
+  return row ? mapNode(row) : null;
+}
+
+export function ensureNode(
+  db: Db,
+  serviceId: number,
+  address: string,
+  meta?: { weight?: number; priority?: number; protocol?: string; port?: number | null },
+): ServiceNode {
+  const existing = findNodeByAddress(db, serviceId, address);
+  if (existing) return existing;
+  const id = db
+    .insert(nodes)
+    .values({
+      service_id: serviceId,
+      address,
+      protocol: meta?.protocol ?? "tcp",
+      port: meta?.port ?? null,
+      weight: meta?.weight ?? 1,
+      priority: meta?.priority ?? 1,
+    })
+    .returning({ id: nodes.id })
+    .get()!.id;
+  return getNode(db, id);
+}
+
+export function createNode(
+  db: Db,
+  serviceId: number,
+  input: {
+    address: string;
+    protocol?: string;
+    port?: number | null;
+    enabled?: boolean;
+    priority?: number;
+    weight?: number;
+    health_check_id?: number | null;
+  },
+): ServiceNode {
+  getService(db, serviceId);
+  const existing = findNodeByAddress(db, serviceId, input.address);
+  if (existing) {
+    throw new ConflictError(`node ${input.address} already exists`);
+  }
+  const id = db
+    .insert(nodes)
+    .values({
+      service_id: serviceId,
+      address: input.address,
+      protocol: input.protocol ?? "tcp",
+      port: input.port ?? null,
+      enabled: input.enabled ?? true,
+      priority: input.priority ?? 1,
+      weight: input.weight ?? 1,
+      health_check_id: input.health_check_id ?? null,
+    })
+    .returning({ id: nodes.id })
+    .get()!.id;
+  insertServiceIpIfMissing(db, serviceId, input.address);
+  return getNode(db, id);
+}
+
+export function updateNode(
+  db: Db,
+  id: number,
+  patch: Partial<{
+    address: string;
+    protocol: string;
+    port: number | null;
+    enabled: boolean;
+    priority: number;
+    weight: number;
+    health_check_id: number | null;
+    health_status: string;
+    consecutive_failures: number;
+    consecutive_successes: number;
+    last_check_at: string | null;
+    last_failure_reason: string | null;
+  }>,
+): ServiceNode {
+  const current = getNode(db, id);
+  const update: Record<string, unknown> = { updated_at: sql`datetime('now')` };
+  for (const [key, value] of Object.entries(patch)) {
+    if (value !== undefined) update[key] = value;
+  }
+  db.update(nodes).set(update).where(eq(nodes.id, id)).run();
+  if (patch.address && patch.address !== current.address) {
+    db.delete(serviceIps)
+      .where(
+        and(
+          eq(serviceIps.service_id, current.service_id),
+          eq(serviceIps.ip, current.address),
+        ),
+      )
+      .run();
+    insertServiceIpIfMissing(db, current.service_id, patch.address);
+  }
+  return getNode(db, id);
+}
+
+export function deleteNode(db: Db, id: number): void {
+  const current = getNode(db, id);
+  db.delete(nodes).where(eq(nodes.id, id)).run();
+  db.delete(serviceIps)
+    .where(
+      and(
+        eq(serviceIps.service_id, current.service_id),
+        eq(serviceIps.ip, current.address),
+      ),
+    )
+    .run();
+}
+
+function mapHealthCheck(row: typeof healthChecks.$inferSelect): OriginHealthCheck {
+  return {
+    id: row.id,
+    provider: row.provider as OriginHealthCheck["provider"],
+    cf_healthcheck_id: row.cf_healthcheck_id,
+    cf_zone_id: row.cf_zone_id,
+    name: row.name,
+    protocol: row.protocol,
+    path: row.path,
+    method: row.method,
+    timeout: row.timeout,
+    interval_sec: row.interval_sec,
+    retries: row.retries,
+    expected_status: row.expected_status,
+    consecutive_fails: row.consecutive_fails,
+    consecutive_successes: row.consecutive_successes,
+    suspended: Boolean(row.suspended),
+    created_at: row.created_at,
+    updated_at: row.updated_at,
+  };
+}
+
+export function listHealthChecks(db: Db): OriginHealthCheck[] {
+  return db.select().from(healthChecks).all().map(mapHealthCheck);
+}
+
+export function getHealthCheck(db: Db, id: number): OriginHealthCheck {
+  const row = db.select().from(healthChecks).where(eq(healthChecks.id, id)).get();
+  if (!row) throw new NotFoundError(`health check ${id}`);
+  return mapHealthCheck(row);
+}
+
+export function findHealthCheckByCfId(
+  db: Db,
+  cfId: string,
+): OriginHealthCheck | null {
+  const row = db
+    .select()
+    .from(healthChecks)
+    .where(eq(healthChecks.cf_healthcheck_id, cfId))
+    .get();
+  return row ? mapHealthCheck(row) : null;
+}
+
+export function createHealthCheck(
+  db: Db,
+  input: {
+    provider: string;
+    name: string;
+    cf_healthcheck_id?: string | null;
+    cf_zone_id?: string | null;
+    protocol?: string;
+    path?: string | null;
+    method?: string | null;
+    timeout?: number;
+    interval_sec?: number;
+    retries?: number;
+    expected_status?: number | null;
+    consecutive_fails?: number;
+    consecutive_successes?: number;
+    suspended?: boolean;
+  },
+): OriginHealthCheck {
+  const id = db
+    .insert(healthChecks)
+    .values({
+      provider: input.provider,
+      name: input.name,
+      cf_healthcheck_id: input.cf_healthcheck_id ?? null,
+      cf_zone_id: input.cf_zone_id ?? null,
+      protocol: input.protocol ?? "tcp",
+      path: input.path ?? null,
+      method: input.method ?? null,
+      timeout: input.timeout ?? 5,
+      interval_sec: input.interval_sec ?? 30,
+      retries: input.retries ?? 2,
+      expected_status: input.expected_status ?? null,
+      consecutive_fails: input.consecutive_fails ?? 2,
+      consecutive_successes: input.consecutive_successes ?? 2,
+      suspended: input.suspended ?? false,
+    })
+    .returning({ id: healthChecks.id })
+    .get()!.id;
+  return getHealthCheck(db, id);
+}
+
+export function updateHealthCheck(
+  db: Db,
+  id: number,
+  patch: Partial<{
+    provider: string;
+    name: string;
+    cf_healthcheck_id: string | null;
+    cf_zone_id: string | null;
+    protocol: string;
+    path: string | null;
+    method: string | null;
+    timeout: number;
+    interval_sec: number;
+    retries: number;
+    expected_status: number | null;
+    consecutive_fails: number;
+    consecutive_successes: number;
+    suspended: boolean;
+  }>,
+): OriginHealthCheck {
+  getHealthCheck(db, id);
+  const update: Record<string, unknown> = { updated_at: sql`datetime('now')` };
+  for (const [key, value] of Object.entries(patch)) {
+    if (value !== undefined) update[key] = value;
+  }
+  db.update(healthChecks).set(update).where(eq(healthChecks.id, id)).run();
+  return getHealthCheck(db, id);
+}
+
+export function deleteHealthCheck(db: Db, id: number): void {
+  const result = db.delete(healthChecks).where(eq(healthChecks.id, id)).run();
+  if (result.changes === 0) throw new NotFoundError(`health check ${id}`);
+}
+
+export function bumpBindingVersion(db: Db, bindingId: number, expected?: number): number {
+  const binding = getBinding(db, bindingId);
+  if (expected != null && binding.operation_version !== expected) {
+    throw new ConflictError(`binding ${bindingId} version conflict`);
+  }
+  const next = (binding.operation_version ?? 0) + 1;
+  db.update(serviceBindings)
+    .set({
+      operation_version: next,
+      updated_at: sql`datetime('now')`,
+    })
+    .where(eq(serviceBindings.id, bindingId))
+    .run();
+  return next;
+}
+
+export function setBindingRoutingStrategy(
+  db: Db,
+  bindingId: number,
+  strategy: LbMode,
+): void {
+  db.update(serviceBindings)
+    .set({
+      routing_strategy: strategy,
+      lb_mode: strategy,
+      updated_at: sql`datetime('now')`,
+    })
+    .where(eq(serviceBindings.id, bindingId))
+    .run();
+}
+
+export function listAllNodes(db: Db): ServiceNode[] {
+  return db.select().from(nodes).all().map(mapNode);
+}
+
+export function updateBindingDomain(
+  db: Db,
+  bindingId: number,
+  domainId: number,
+  hostname: string,
+): void {
+  db.update(serviceBindings)
+    .set({
+      domain_id: domainId,
+      hostname,
+      updated_at: sql`datetime('now')`,
+    })
+    .where(eq(serviceBindings.id, bindingId))
+    .run();
+}
+
+export function listBindingNodes(db: Db, bindingId: number): ServiceNode[] {
+  return db
+    .select({ node: nodes })
+    .from(bindingNodes)
+    .innerJoin(nodes, eq(bindingNodes.node_id, nodes.id))
+    .where(eq(bindingNodes.binding_id, bindingId))
+    .all()
+    .map((row) => mapNode(row.node));
 }
 
 // --- Service Binding IPs ---
@@ -980,9 +1352,13 @@ export function replaceBindingIpsWithMeta(
   bindingId: number,
   entries: BindingIpMeta[],
 ): void {
+  const binding = getBinding(db, bindingId);
   db.delete(serviceBindingIps)
     .where(eq(serviceBindingIps.binding_id, bindingId))
     .run();
+  db.delete(bindingNodes)
+    .where(eq(bindingNodes.binding_id, bindingId))
+    .run();
   for (const entry of entries) {
     db.insert(serviceBindingIps)
       .values({
@@ -992,6 +1368,18 @@ export function replaceBindingIpsWithMeta(
         priority: entry.priority,
       })
       .run();
+    const node = ensureNode(db, binding.service_id, entry.ip, {
+      weight: entry.weight,
+      priority: entry.priority,
+    });
+    db.insert(bindingNodes)
+      .values({
+        binding_id: bindingId,
+        node_id: node.id,
+        weight: entry.weight,
+        priority: entry.priority,
+      })
+      .run();
   }
 }
 
@@ -1015,7 +1403,10 @@ export function updateBindingLbConfig(
   const update: Record<string, unknown> = {
     updated_at: sql`datetime('now')`,
   };
-  if (patch.lb_mode !== undefined) update.lb_mode = patch.lb_mode;
+  if (patch.lb_mode !== undefined) {
+    update.lb_mode = patch.lb_mode;
+    update.routing_strategy = patch.lb_mode;
+  }
   if (patch.health_check_enabled !== undefined)
     update.health_check_enabled = patch.health_check_enabled;
   if (patch.health_check_type !== undefined)
@@ -1721,7 +2112,7 @@ export function getIpHealthStatusRow(
 ): IpHealthStatus | null {
   const rows = db.all<IpHealthStatus>(sql`
     SELECT scope, ref_id, ip, status, latency_ms, consecutive_failures,
-      last_checked_at, last_error
+      consecutive_successes, last_checked_at, last_error
     FROM ip_health_status
     WHERE scope = ${scope} AND ref_id = ${refId} AND ip = ${ip}
     LIMIT 1
@@ -1738,17 +2129,19 @@ export function upsertIpHealthStatus(
   latencyMs: number | null,
   consecutiveFailures: number,
   lastError: string | null,
+  consecutiveSuccesses = 0,
 ): void {
   db.run(sql`
     INSERT INTO ip_health_status
       (scope, ref_id, ip, status, latency_ms, consecutive_failures,
-       last_checked_at, last_error, created_at, updated_at)
+       consecutive_successes, last_checked_at, last_error, created_at, updated_at)
     VALUES (${scope}, ${refId}, ${ip}, ${status}, ${latencyMs}, ${consecutiveFailures},
-      datetime('now'), ${lastError}, datetime('now'), datetime('now'))
+      ${consecutiveSuccesses}, datetime('now'), ${lastError}, datetime('now'), datetime('now'))
     ON CONFLICT(scope, ref_id, ip) DO UPDATE SET
       status = excluded.status,
       latency_ms = excluded.latency_ms,
       consecutive_failures = excluded.consecutive_failures,
+      consecutive_successes = excluded.consecutive_successes,
       last_checked_at = excluded.last_checked_at,
       last_error = excluded.last_error,
       updated_at = datetime('now')

packages/db/src/schema.ts

@@ -165,6 +165,8 @@ export const serviceBindings = sqliteTable(
     })
       .notNull()
       .default(false),
+    routing_strategy: text("routing_strategy").notNull().default("round_robin"),
+    operation_version: integer("operation_version").notNull().default(0),
     created_at: text("created_at")
       .notNull()
       .default(sql`datetime('now')`),
@@ -181,6 +183,76 @@ export const serviceBindings = sqliteTable(
   ],
 );
 
+export const healthChecks = sqliteTable("health_checks", {
+  id: integer("id").primaryKey({ autoIncrement: true }),
+  provider: text("provider").notNull().default("local"),
+  cf_healthcheck_id: text("cf_healthcheck_id"),
+  cf_zone_id: text("cf_zone_id"),
+  name: text("name").notNull(),
+  protocol: text("protocol").notNull().default("tcp"),
+  path: text("path"),
+  method: text("method"),
+  timeout: integer("timeout").notNull().default(5),
+  interval_sec: integer("interval_sec").notNull().default(30),
+  retries: integer("retries").notNull().default(2),
+  expected_status: integer("expected_status"),
+  consecutive_fails: integer("consecutive_fails").notNull().default(2),
+  consecutive_successes: integer("consecutive_successes").notNull().default(2),
+  suspended: integer("suspended", { mode: "boolean" }).notNull().default(false),
+  created_at: text("created_at")
+    .notNull()
+    .default(sql`datetime('now')`),
+  updated_at: text("updated_at")
+    .notNull()
+    .default(sql`datetime('now')`),
+});
+
+export const nodes = sqliteTable(
+  "nodes",
+  {
+    id: integer("id").primaryKey({ autoIncrement: true }),
+    service_id: integer("service_id")
+      .notNull()
+      .references(() => services.id, { onDelete: "cascade" }),
+    address: text("address").notNull(),
+    protocol: text("protocol").notNull().default("tcp"),
+    port: integer("port"),
+    enabled: integer("enabled", { mode: "boolean" }).notNull().default(true),
+    priority: integer("priority").notNull().default(1),
+    weight: integer("weight").notNull().default(1),
+    health_status: text("health_status").notNull().default("unknown"),
+    health_check_id: integer("health_check_id").references(() => healthChecks.id, {
+      onDelete: "set null",
+    }),
+    consecutive_failures: integer("consecutive_failures").notNull().default(0),
+    consecutive_successes: integer("consecutive_successes").notNull().default(0),
+    last_check_at: text("last_check_at"),
+    last_failure_reason: text("last_failure_reason"),
+    created_at: text("created_at")
+      .notNull()
+      .default(sql`datetime('now')`),
+    updated_at: text("updated_at")
+      .notNull()
+      .default(sql`datetime('now')`),
+  },
+  (t) => [unique("nodes_service_address").on(t.service_id, t.address)],
+);
+
+export const bindingNodes = sqliteTable(
+  "binding_nodes",
+  {
+    binding_id: integer("binding_id")
+      .notNull()
+      .references(() => serviceBindings.id, { onDelete: "cascade" }),
+    node_id: integer("node_id")
+      .notNull()
+      .references(() => nodes.id, { onDelete: "cascade" }),
+    weight: integer("weight").notNull().default(1),
+    priority: integer("priority").notNull().default(1),
+  },
+  (t) => [primaryKey({ columns: [t.binding_id, t.node_id] })],
+);
+
 export const serviceIps = sqliteTable("service_ips", {
   id: integer("id").primaryKey({ autoIncrement: true }),
   service_id: integer("service_id")
@@ -276,6 +348,9 @@ export const ipHealthStatus = sqliteTable(
     consecutive_failures: integer("consecutive_failures")
       .notNull()
       .default(0),
+    consecutive_successes: integer("consecutive_successes")
+      .notNull()
+      .default(0),
     last_checked_at: text("last_checked_at"),
     last_error: text("last_error"),
     created_at: text("created_at")
@@ -399,6 +474,9 @@ export const schema = {
   subdomains,
   dnsRecords,
   serviceBindings,
+  healthChecks,
+  nodes,
+  bindingNodes,
   serviceIps,
   serviceBindingRecords,
   serviceBindingIps,