updates

src/resources/homelab/authentik-server/authentik-server.ts

@@ -215,7 +215,7 @@ class AuthentikServer extends CustomResource<typeof specSchema> {
     });
 
     const gateway = this.#environment.current.gateway;
-    await this.#virtualService.ensure({
+    await this.#virtualService.set({
       metadata: {
         ownerReferences: [this.ref],
       },

src/resources/homelab/environment/environment.ts

@@ -146,7 +146,7 @@ class Environment extends CustomResource<typeof specSchema> {
       },
     });
 
-    await this.#gateway.ensure({
+    await this.#gateway.set({
       metadata: {
         ownerReferences: [this.ref],
       },

src/resources/istio/destination-rule/destination-rule.ts

@@ -1,11 +1,32 @@
 import type { KubernetesObject } from '@kubernetes/client-node';
 import type { K8SDestinationRuleV1 } from 'src/__generated__/resources/K8SDestinationRuleV1.ts';
 
-import { Resource } from '#services/resources/resources.ts';
+import { Resource, ResourceService, type ResourceOptions } from '#services/resources/resources.ts';
+import { CRD } from '#resources/core/crd/crd.ts';
+import { NotReadyError } from '#utils/errors.ts';
 
 class DestinationRule extends Resource<KubernetesObject & K8SDestinationRuleV1> {
   public static readonly apiVersion = 'networking.istio.io/v1';
   public static readonly kind = 'DestinationRule';
+
+  #crd: CRD;
+
+  constructor(options: ResourceOptions<KubernetesObject & K8SDestinationRuleV1>) {
+    super(options);
+    const resourceService = this.services.get(ResourceService);
+    this.#crd = resourceService.get(CRD, 'destinationrules.networking.istio.io');
+  }
+
+  public get hasCRD() {
+    return this.#crd.exists;
+  }
+
+  public set = async (manifest: KubernetesObject & K8SDestinationRuleV1) => {
+    if (!this.hasCRD) {
+      throw new NotReadyError('CRD is not installed');
+    }
+    await this.ensure(manifest);
+  };
 }
 
 export { DestinationRule };

src/resources/istio/gateway/gateway.ts

@@ -1,11 +1,37 @@
 import type { KubernetesObject } from '@kubernetes/client-node';
 import type { K8SGatewayV1 } from 'src/__generated__/resources/K8SGatewayV1.ts';
 
-import { Resource } from '#services/resources/resources.ts';
+import { Resource, ResourceService, type ResourceOptions } from '#services/resources/resources.ts';
+import { CRD } from '#resources/core/crd/crd.ts';
+import { NotReadyError } from '#utils/errors.ts';
 
 class Gateway extends Resource<KubernetesObject & K8SGatewayV1> {
   public static readonly apiVersion = 'networking.istio.io/v1';
   public static readonly kind = 'Gateway';
+
+  #crd: CRD;
+
+  constructor(options: ResourceOptions<KubernetesObject & K8SGatewayV1>) {
+    super(options);
+    const resourceService = this.services.get(ResourceService);
+    this.#crd = resourceService.get(CRD, 'gateways.networking.istio.io');
+    this.on('changed', this.#handleUpdate);
+  }
+
+  #handleUpdate = async () => {
+    this.emit('changed');
+  };
+
+  public get hasCRD() {
+    return this.#crd.exists;
+  }
+
+  public set = async (manifest: KubernetesObject & K8SGatewayV1) => {
+    if (!this.hasCRD) {
+      throw new NotReadyError('CRD is not installed');
+    }
+    await this.ensure(manifest);
+  };
 }
 
 export { Gateway };

src/resources/istio/virtual-service/virtual-service.ts

@@ -1,11 +1,32 @@
 import type { KubernetesObject } from '@kubernetes/client-node';
 import type { K8SVirtualServiceV1 } from 'src/__generated__/resources/K8SVirtualServiceV1.ts';
 
-import { Resource } from '#services/resources/resources.ts';
+import { Resource, ResourceService, type ResourceOptions } from '#services/resources/resources.ts';
+import { CRD } from '#resources/core/crd/crd.ts';
+import { NotReadyError } from '#utils/errors.ts';
 
 class VirtualService extends Resource<KubernetesObject & K8SVirtualServiceV1> {
   public static readonly apiVersion = 'networking.istio.io/v1';
   public static readonly kind = 'VirtualService';
+
+  #crd: CRD;
+
+  constructor(options: ResourceOptions<KubernetesObject & K8SVirtualServiceV1>) {
+    super(options);
+    const resourceService = this.services.get(ResourceService);
+    this.#crd = resourceService.get(CRD, 'virtualservices.networking.istio.io');
+  }
+
+  public get hasCRD() {
+    return this.#crd.exists;
+  }
+
+  public set = async (manifest: KubernetesObject & K8SVirtualServiceV1) => {
+    if (!this.hasCRD) {
+      throw new NotReadyError('CRD is not installed');
+    }
+    await this.ensure(manifest);
+  };
 }
 
 export { VirtualService };

src/services/resources/resource/resource.custom.ts

@@ -1,9 +1,11 @@
 import { z, type ZodType } from 'zod';
-import type { KubernetesObject } from '@kubernetes/client-node';
+import { type KubernetesObject } from '@kubernetes/client-node';
 
 import { Resource, type ResourceOptions } from './resource.ts';
 
 import { API_VERSION } from '#utils/consts.ts';
+import { CoalescingQueued } from '#utils/queues.ts';
+import { NotReadyError } from '#utils/errors.ts';
 
 const customResourceStatusSchema = z.object({
   observedGeneration: z.number().optional(),
@@ -26,9 +28,69 @@ const customResourceStatusSchema = z.object({
 
 type CustomResourceOptions<TSpec extends ZodType> = ResourceOptions<KubernetesObject & { spec: z.infer<TSpec> }>;
 
-class CustomResource<TSpec extends ZodType> extends Resource<KubernetesObject & { spec: z.infer<TSpec> }> {
+class CustomResource<TSpec extends ZodType> extends Resource<
+  KubernetesObject & { spec: z.infer<TSpec>; status?: z.infer<typeof customResourceStatusSchema> }
+> {
   public static readonly apiVersion = API_VERSION;
   public static readonly status = customResourceStatusSchema;
+
+  #reconcileQueue: CoalescingQueued<void>;
+
+  constructor(options: CustomResourceOptions<TSpec>) {
+    super(options);
+    this.#reconcileQueue = new CoalescingQueued({
+      action: async () => {
+        try {
+          if (!this.exists || !this.manifest?.metadata?.deletionTimestamp) {
+            return;
+          }
+          this.services.log.debug('Reconciling', {
+            apiVersion: this.apiVersion,
+            kind: this.kind,
+            namespace: this.namespace,
+            name: this.name,
+          });
+          await this.reconcile?.();
+        } catch (err) {
+          if (err instanceof NotReadyError) {
+            console.error(err);
+          } else {
+            throw err;
+          }
+        }
+      },
+    });
+    this.on('changed', this.#handleUpdate);
+  }
+
+  public get isSeen() {
+    return this.metadata?.generation === this.status?.observedGeneration;
+  }
+
+  #handleUpdate = async () => {
+    if (this.isSeen) {
+      return;
+    }
+    return await this.queueReconcile();
+  };
+
+  public reconcile?: () => Promise<void>;
+  public queueReconcile = () => {
+    return this.#reconcileQueue.run();
+  };
+
+  public markSeen = async () => {
+    if (this.isSeen) {
+      return;
+    }
+    await this.patchStatus({
+      observedGeneration: this.metadata?.generation,
+    });
+  };
+
+  public patchStatus = async (status: Partial<z.infer<typeof customResourceStatusSchema>>) => {
+    this.patch({ status } as ExpectedAny);
+  };
 }
 
 export { CustomResource, type CustomResourceOptions };

src/services/resources/resource/resource.ts

@@ -7,8 +7,6 @@ import { Queue } from '../../queue/queue.ts';
 import { K8sService } from '../../k8s/k8s.ts';
 import { isDeepSubset } from '../../../utils/objects.ts';
 
-import { CoalescingQueued } from '#utils/queues.ts';
-
 type ResourceSelector = {
   apiVersion: string;
   kind: string;
@@ -29,7 +27,6 @@ type ResourceEvents = {
 class Resource<T extends KubernetesObject> extends EventEmitter<ResourceEvents> {
   #manifest?: T;
   #queue: Queue;
-  #reconcileQueue: CoalescingQueued<void>;
   #options: ResourceOptions<T>;
 
   constructor(options: ResourceOptions<T>) {
@@ -37,27 +34,8 @@ class Resource<T extends KubernetesObject> extends EventEmitter<ResourceEvents>
     this.#options = options;
     this.#manifest = options.manifest;
     this.#queue = new Queue({ concurrency: 1 });
-    this.#reconcileQueue = new CoalescingQueued({
-      action: async () => {
-        try {
-          if (!this.exists || !this.reconcile) {
-            return;
-          }
-          console.log('Reconcileing', this.apiVersion, this.kind, this.namespace, this.name);
-          await this.reconcile?.();
-        } catch (err) {
-          console.error(err);
-        }
-      },
-    });
-    this.on('changed', this.queueReconcile);
   }
 
-  public reconcile?: () => Promise<void>;
-  public queueReconcile = () => {
-    return this.#reconcileQueue.run();
-  };
-
   public get services() {
     return this.#options.services;
   }

src/utils/errors.ts

@@ -0,0 +1,14 @@
+class NotReadyError extends Error {
+  #reason?: string;
+
+  constructor(reason?: string, message?: string) {
+    super(message || reason || 'Resource is not ready');
+    this.#reason = reason;
+  }
+
+  get reason() {
+    return this.#reason;
+  }
+}
+
+export { NotReadyError };